<commit_before>/**
 * Copyright (C) 2009-2013 FoundationDB, LLC
 *
 * This program is free software: you can redistribute it and/or modify
 * it under the terms of the GNU Affero General Public License as published by
 * the Free Software Foundation, either version 3 of the License, or
 * (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU Affero General Public License for more details.
 *
 * You should have received a copy of the GNU Affero General Public License
 * along with this program.  If not, see <http://www.gnu.org/licenses/>.
 */

package com.foundationdb.server.store;

import com.foundationdb.ais.model.AkibanInformationSchema;
import com.foundationdb.ais.model.Columnar;
import com.foundationdb.ais.model.NameGenerator;
import com.foundationdb.ais.model.Routine;
import com.foundationdb.ais.model.SQLJJar;
import com.foundationdb.ais.model.Sequence;
import com.foundationdb.ais.model.SynchronizedNameGenerator;
import com.foundationdb.ais.model.Table;
import com.foundationdb.ais.model.TableName;
import com.foundationdb.ais.model.aisb2.AISBBasedBuilder;
import com.foundationdb.ais.model.aisb2.NewAISBuilder;
import com.foundationdb.ais.model.validation.AISValidations;
import com.foundationdb.ais.protobuf.ProtobufReader;
import com.foundationdb.ais.protobuf.ProtobufWriter;
import com.foundationdb.ais.protobuf.ProtobufWriter.WriteSelector;
import com.foundationdb.ais.util.UuidAssigner;
import com.foundationdb.qp.memoryadapter.BasicFactoryBase;
import com.foundationdb.qp.memoryadapter.MemoryAdapter;
import com.foundationdb.qp.memoryadapter.MemoryGroupCursor;
import com.foundationdb.qp.row.ValuesRow;
import com.foundationdb.qp.row.Row;
import com.foundationdb.qp.rowtype.RowType;
import com.foundationdb.server.PersistitAccumulatorTableStatusCache;
import com.foundationdb.server.error.AkibanInternalException;
import com.foundationdb.server.rowdata.RowDefCache;
import com.foundationdb.server.service.config.ConfigurationService;
import com.foundationdb.server.service.security.SecurityService;
import com.foundationdb.server.service.session.Session;
import com.foundationdb.server.service.session.SessionService;
import com.foundationdb.server.service.transaction.TransactionService;
import com.foundationdb.server.service.tree.TreeLink;
import com.foundationdb.server.service.tree.TreeService;
import com.foundationdb.server.service.tree.TreeVisitor;
import com.foundationdb.server.store.TableChanges.ChangeSet;
import com.foundationdb.server.store.format.PersistitStorageDescription;
import com.foundationdb.server.store.format.PersistitStorageFormatRegistry;
import com.foundationdb.server.util.ReadWriteMap;
import com.google.inject.Inject;
import com.persistit.Accumulator;
import com.persistit.Exchange;
import com.persistit.Key;
import com.persistit.KeyFilter;
import com.persistit.exception.PersistitException;
import com.persistit.exception.RollbackException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.nio.ByteBuffer;
import java.util.Collection;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Callable;
import java.util.concurrent.DelayQueue;
import java.util.concurrent.Delayed;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;

import static com.foundationdb.qp.storeadapter.PersistitAdapter.wrapPersistitException;
import static com.foundationdb.server.service.transaction.TransactionService.Callback;
import static com.foundationdb.server.service.transaction.TransactionService.CallbackType;
import static com.foundationdb.server.service.tree.TreeService.SCHEMA_TREE_NAME;

/**
 * Version 1.9.2 - 2013-10
 * <pre>
 *     - Data format version for non-SchemaManager (i.e. row/index) contents
 *     - Metadata format version for SchemaManager contents
 *     - Accumulator, at {@link #ACCUMULATOR_INDEX_SCHEMA_GEN}, for AIS generation
 *     - Serialized Protobuf per schema, via {@link ProtobufWriter}
 *     - Removed trees stored, with id and timestamp, for cleanup at next start
 *     - In-progress DDL, identified by session ID, stores two chunks of data:
 *         - Each affected table gets a key mapping to the DDL ID
 *         - New serialized Protobuf stored for each affected schema
 *
 *     {@link TreeService#SCHEMA_TREE_NAME}
 *         Accumulator[0]                           =>  Seq (generation)
 *         Accumulator[1]                           =>  Seq (online IDs)
 *         "metaVersion"                            =>  long
 *         "dataVersion"                            =>  long
 *         "delayed",(long)id,(long)ts              =>  "schema","tree"
 *         ...
 *         "protobuf","schema"                      =>  byte[] (AIS Protobuf)
 *         ...
 *         "protobufMem"                            =>  byte[] (AIS Protobuf)
 *         ...
 *         "online",(long)id                        =>  long (generation)
 *         ...
 *         "online",(long)id,"change",(int)tid      =>  byte[] (ChangeSet Protobuf)
 *         ...
 *         "online",(long)id,"protobuf","schema"    =>  byte[] (AIS Protobuf)
 *         ...
 * </pre>
 *
 * Version 1.2.1 - 2012-05
 * <pre>
 *     - Accumulator, at {@link #ACCUMULATOR_INDEX_SCHEMA_GEN}, for AIS generation
 *     - Serialized Protobuf per schema, via {@link ProtobufWriter}
 *     - Single "version" stored with every schema entry
 *     - Removed trees stored, with id and timestamp, for cleanup at next start
 *
 *     {@link TreeService#SCHEMA_TREE_NAME}
 *         Accumulator[0]                       =>  Seq
 *         "delayedTree",(long)id,(long)ts      =>  "schema","treeName"
 *         ...
 *         "by","PBAIS",(int)ver,"schema"       =>  byte[]
 *         ...
 *         "by","PBMEMAIS"                      =>  byte[]
 * </pre>
 *
 * Version 0.4 - 2011-05
 * <pre>
 *     - Single k/v pair containing the entire serialized AIS, via the
 *       deleted com.foundationdb.ais.metamodel.io.MessageTarget.
 *
 *     {@link TreeService#SCHEMA_TREE_NAME}
 *         "byAIS"      =>  byte[]
 * </pre>
 */
public class PersistitStoreSchemaManager extends AbstractSchemaManager {
    private static enum GenValue { NEW, SNAPSHOT }
    private static enum GenMap { PUT_NEW, NO_PUT }

    private static class SharedAIS {
        public final AtomicInteger refCount;
        public final AkibanInformationSchema ais;

        public SharedAIS(AkibanInformationSchema ais) {
            this.refCount = new AtomicInteger(0);
            this.ais = ais;
        }

        public int acquire() {
            return refCount.incrementAndGet();
        }

        public int release() {
            int count = refCount.decrementAndGet();
            assert count >= 0 : count;
            return count;
        }

        public int shareCount() {
            return refCount.get();
        }

        public boolean isShared() {
            return refCount.get() > 0;
        }

        @Override
        public String toString() {
            return "Shared[" + refCount.get() + "]" + ais;
        }
    }

    private static class AISAndTimestamp {
        public final SharedAIS sAIS;
        public final long timestamp;

        public AISAndTimestamp(SharedAIS sAIS, long timestamp) {
            this.sAIS = sAIS;
            this.timestamp = timestamp;
        }

        public boolean isUsableForStartTime(long startTime) {
            return startTime > timestamp;
        }

        @Override
        public String toString() {
            return sAIS + ":" + timestamp;
        }
    }

    private static final long CURRENT_DATA_VERSION = 1;
    private static final long CURRENT_META_VERSION = 1;
    private static final String S_K_META_VERSION = "metaVersion";
    private static final String S_K_DATA_VERSION = "dataVersion";
    private static final String S_K_DELAYED = "delayed";
    private static final String S_K_PROTOBUF = "protobuf";
    private static final String S_K_PROTOBUF_MEM = "protobufMem";
    private static final String S_K_ONLINE = "online";
    private static final String S_K_CHANGE = "change";
    private static final int ACCUMULATOR_INDEX_SCHEMA_GEN = 0;
    private static final int ACCUMULATOR_INDEX_ONLINE_ID = 1;

    /** Used when a consistent volume is required (e.g. accumulator) no matter what. */
    private static final String SCHEMA_TREE_INTERNAL_SCHEMA = "pssm";
    private static final Session.Key<SharedAIS> SESSION_SAIS_KEY = Session.Key.named("PSSM_SAIS");

    /**
     * <p>This is used as unusable cache identifier, count for outstanding DDLs, and sync object for updating
     * the latest cache value.</p>
     * <p>Why are all needed? Consider the simple case two concurrent DDLs, A and B. Both are have pre-committed
     * and cleared the cache. If A commits, the cache is updated, and then B commits, any reader using the cache
     * based on timestamp alone will get a stale snapshot. So, the cache can *only* be updated when there are no
     * more outstanding.</p>
     */
    private static final AISAndTimestamp CACHE_SENTINEL = new AISAndTimestamp(new SharedAIS(null), Long.MAX_VALUE);

    private static final Logger LOG = LoggerFactory.getLogger(PersistitStoreSchemaManager.class);

    private final TreeService treeService;
    private RowDefCache rowDefCache;
    private NameGenerator nameGenerator;
    private AtomicLong delayedTreeIDGenerator;
    private ReadWriteMap<Long,SharedAIS> aisMap;
    private volatile AISAndTimestamp latestAISCache;
    private TransactionService.Callback clearLatestCacheCallback;
    private TransactionService.Callback enqueueClearAndUpdateCallback;
    private BlockingQueue<QueueTask> taskQueue;
    private Thread queueConsumer;
    private AtomicInteger aisCacheMissCount;
    private AtomicInteger loadAISFromStorageCount;
    private AtomicInteger delayedTreeCount;

    @Inject
    public PersistitStoreSchemaManager(ConfigurationService config, SessionService sessionService,
                                       TreeService treeService, TransactionService txnService) {
        super(config, sessionService, txnService, new PersistitStorageFormatRegistry());
        this.treeService = treeService;
    }

    @Override
    protected NameGenerator getNameGenerator(Session session) {
        return nameGenerator;
    }

    @Override
    public void setSecurityService(SecurityService securityService) {
        this.securityService = securityService;
    }

    @Override
    public void addOnlineChangeSet(Session session, ChangeSet changeSet) {
        OnlineSession onlineSession = getOnlineSession(session, true);
        Exchange ex = getInternalExchange(session);
        try {
            ex.clear().append(S_K_ONLINE).append(onlineSession.id).append(S_K_CHANGE).append(changeSet.getTableId());
            ex.getValue().putByteArray(ChangeSetHelper.save(changeSet));
            ex.store();
        } catch(PersistitException | RollbackException e) {
            if(ex != null) {
                treeService.releaseExchange(session, ex);
            }
        }
    }

    @Override
    public AkibanInformationSchema getAis(Session session) {
        AkibanInformationSchema ais = getAISInternal(session).ais;
        OnlineSession onlineSession = getOnlineSession(session, null);
        if(onlineSession != null) {
            AkibanInformationSchema onlineAIS = getOnlineCache(session, ais).onlineToAIS.get(onlineSession.id);
            if(onlineAIS != null) {
                ais = onlineAIS;
            }
        }
        return ais;
    }

    private SharedAIS getAISInternal(Session session) {
        SharedAIS local = session.get(SESSION_SAIS_KEY);
        if(local != null) {
            return local;
        }

        // Latest is a volatile read and that is all that is required. If it is a valid cache, it can only
        // transition to non-valid (pre-committed DDL that doesn't affect out read validity) or the same
        // snapshot with a higher timestamp. See checks in updateLatestAISCache.
        final AISAndTimestamp cached = latestAISCache;
        final long startTimestamp = txnService.getTransactionStartTimestamp(session);
        if(cached.isUsableForStartTime(startTimestamp)) {
            local = cached.sAIS;
        }

        // Couldn't use the cache, do an always accurate accumulator lookup and check in map.
        long generation = 0;
        if(local == null) {
            aisCacheMissCount.incrementAndGet();
            generation = getGenerationSnapshot(session);
            local = aisMap.get(generation);
        }

        // Wasn't in map so need to reload from disk.
        // Should be 1) very rare and 2) fairly quick, so just do it under write lock to avoid duplicate work/entries
        if(local == null) {
            aisMap.claimExclusive();
            try {
                // Double check while while under exclusive
                local = aisMap.get(generation);
                if(local == null) {
                    loadAISFromStorageCount.incrementAndGet();
                    try {
                        local = loadToShared(session, GenValue.SNAPSHOT, GenMap.PUT_NEW);
                        buildRowDefCache(session, local.ais);
                    } catch(PersistitException | RollbackException e) {
                        throw wrapPersistitException(session, e);
                    }
                }
            } finally {
                aisMap.releaseExclusive();
            }
        }

        attachToSession(session, local);
        return local;
    }

    void treeWasRemoved(Session session, String schema, String treeName) {
        LOG.debug("Delaying removal of tree (until next restart): {}", treeName);
        delayedTreeCount.incrementAndGet();
        Exchange ex = schemaTreeExchange(session, schema);
        try {
            long id = delayedTreeIDGenerator.incrementAndGet();
            long timestamp = System.currentTimeMillis();
            ex.clear().append(S_K_DELAYED).append(id).append(timestamp);
            ex.getValue().setStreamMode(true);
            ex.getValue().put(schema);
            ex.getValue().put(treeName);
            ex.getValue().setStreamMode(false);
            ex.store();
        } catch(PersistitException | RollbackException e) {
            throw wrapPersistitException(session, e);
        } finally {
            treeService.releaseExchange(session, ex);
        }
    }

    @Override
    public void start() {
        super.start();

        rowDefCache = new RowDefCache(treeService.getTableStatusCache());
        boolean skipAISUpgrade = Boolean.parseBoolean(config.getProperty(SKIP_AIS_UPGRADE_PROPERTY));

        this.aisMap = ReadWriteMap.wrapNonFair(new HashMap<Long,SharedAIS>());

        final int[] ordinalChangeCount = { 0 };
        final AkibanInformationSchema newAIS;
        try(Session session = sessionService.createSession()) {
            newAIS = txnService.run(session, new Callable<AkibanInformationSchema>() {
                @Override
                public AkibanInformationSchema call() {
                    try {
                        // Unrelated to loading, but fine time to do it
                        cleanupDelayedTrees(session, true);

                        SharedAIS sAIS = loadToShared(session, GenValue.SNAPSHOT, GenMap.PUT_NEW);

                        // Migrate requires RowDefs, but shouldn't generate ordinals (otherwise nulls will be lost)
                        buildRowDefCache(session, sAIS.ais, true);
                        ordinalChangeCount[0] = migrateAccumulatorOrdinals(sAIS.ais);
                        // And generate it fully now that ordinals are definitely set
                        buildRowDefCache(session, sAIS.ais, false);

                        long startTimestamp = txnService.getTransactionStartTimestamp(session);
                        sAIS.acquire(); // So count while in cache is 1
                        latestAISCache = new AISAndTimestamp(sAIS, startTimestamp);
                        return sAIS.ais;
                    } catch(PersistitException e) {
                        throw wrapPersistitException(session, e);
                    }
                }
            });
        }

        this.nameGenerator = SynchronizedNameGenerator.wrap(new PersistitNameGenerator(newAIS));
        this.delayedTreeIDGenerator = new AtomicLong();
        for(Table table : newAIS.getTables().values()) {
            // Note: table.getVersion may be null (pre-1.4.3 volumes)
            tableVersionMap.put(table.getTableId(), table.getVersion());
        }

        this.taskQueue = new DelayQueue<>();
        this.queueConsumer = new Thread(new QueueConsumer(this.taskQueue), "PSSM_QUEUE");
        this.queueConsumer.start();

        this.clearLatestCacheCallback = new Callback() {
            @Override
            public void run(Session session, long timestamp) {
                updateLatestAISCache(CACHE_SENTINEL);
            }
        };
        this.enqueueClearAndUpdateCallback = new Callback() {
            @Override
            public void run(Session session, long timestamp) {
                taskQueue.add(new UpdateLatestCacheTask(0));
                taskQueue.add(new ClearAISMapTask(0, 10000));
            }
        };

        this.aisCacheMissCount = new AtomicInteger(0);
        this.loadAISFromStorageCount = new AtomicInteger(0);
        this.delayedTreeCount = new AtomicInteger(0);

        if (!skipAISUpgrade) {
            final AkibanInformationSchema upgradeAIS = aisCloner.clone(newAIS);
            UuidAssigner uuidAssigner = new UuidAssigner();
            upgradeAIS.visit(uuidAssigner);
            if(uuidAssigner.assignedAny() || ordinalChangeCount[0] > 0) {
                try(Session session = sessionService.createSession()) {
                    txnService.run(session, new Runnable() {
                        @Override
                        public void run() {
                            storedAISChange(session, upgradeAIS, upgradeAIS.getSchemas().keySet());
                        }
                    });
                }
            }
        }
        // else LOG.warn("Skipping AIS upgrade");

        registerSummaryTable();
    }

    @Override
    public void stop() {
        stopConsumer();
        this.delayedTreeIDGenerator = null;
        this.aisMap = null;
        this.nameGenerator = null;
        this.latestAISCache = null;
        this.clearLatestCacheCallback = null;
        this.enqueueClearAndUpdateCallback = null;
        this.taskQueue.clear();
        this.taskQueue = null;
        this.queueConsumer = null;
        this.aisCacheMissCount = null;
        this.loadAISFromStorageCount = null;
        this.delayedTreeCount = null;
        CACHE_SENTINEL.sAIS.refCount.set(0);
        super.stop();
    }

    @Override
    public void crash() {
        stop();
    }

    private void stopConsumer() {
        final long endNanoTime = System.nanoTime() + TimeUnit.SECONDS.toNanos(5);
        if(!taskQueue.offer(new StopTask(0))) {
            // DelayedQueue is unbounded, should never happen
            LOG.error("Could not offer StopTask");
        }
        for(;;) {
            long remaining = endNanoTime - System.nanoTime();
            if(remaining <= 0 || !queueConsumer.isAlive()) {
                break;
            }
            try {
                queueConsumer.join(100);
            } catch(InterruptedException e) {
                LOG.warn("Interrupted while trying to stop QueueConsumer");
            }
        }
    }

    private SharedAIS loadToShared(Session session, GenValue genValue, GenMap genMap) throws PersistitException {
        ProtobufReader reader = new ProtobufReader(storageFormatRegistry, new AkibanInformationSchema());
        loadFromStorage(session, reader);
        AkibanInformationSchema newAIS = finishReader(reader);
        return createValidatedShared(session, newAIS, genValue, genMap);
    }

    private void loadFromStorage(Session session, final ProtobufReader reader) throws PersistitException {
        treeService.visitStorage(session, new TreeVisitor() {
            @Override
            public void visit(Exchange ex) throws PersistitException {
                loadPrimaryProtobuf(ex, reader, null);
            }},
            SCHEMA_TREE_NAME
        );
    }

    private void loadProtobufChildren(Exchange ex, ProtobufReader reader, Collection<String> skipSchemas) throws PersistitException {
        ex.append(Key.BEFORE);
        while(ex.next()) {
            if((skipSchemas != null) && skipSchemas.contains(ex.getKey().indexTo(-1).decodeString())) {
                continue;
            }
            byte[] storedAIS = ex.getValue().getByteArray();
            ByteBuffer buffer = ByteBuffer.wrap(storedAIS);
            reader.loadBuffer(buffer);
        }
    }

    private void loadPrimaryProtobuf(Exchange ex, ProtobufReader reader, Collection<String> skipSchemas) throws PersistitException {
        ex.clear().append(S_K_META_VERSION).fetch();
        if(!ex.getValue().isDefined()) {
            // Can only be empty if there is no data here
            ex.clear();
            if(ex.hasNext()) {
                throw new AkibanInternalException("Unsupported data volume (no metadata version key)");
            } else {
                // No data in this volume at all.
                return;
            }
        }
        long metaVersion = ex.getValue().getLong();
        if(CURRENT_META_VERSION != metaVersion) {
            throw new AkibanInternalException(String.format("Unsupported data volume meta version %d, current %d",
                                                            metaVersion, CURRENT_META_VERSION));
        }

        ex.clear().append(S_K_DATA_VERSION).fetch();
        assert ex.getValue().isDefined() : "No data version";
        long dataVersion = ex.getValue().getLong();
        if(CURRENT_META_VERSION != dataVersion) {
            throw new AkibanInternalException(String.format("Unsupported data volume data version %d, current %d",
                                                            dataVersion, CURRENT_DATA_VERSION));
        }

        ex.clear().append(S_K_PROTOBUF);
        loadProtobufChildren(ex, reader, skipSchemas);

        ex.clear().append(S_K_PROTOBUF_MEM).fetch();
        if(ex.getValue().isDefined()) {
            byte[] storedAIS = ex.getValue().getByteArray();
            ByteBuffer buffer = ByteBuffer.wrap(storedAIS);
            reader.loadBuffer(buffer);
        }
    }
<commit_msg>Bump the DATA_VERSION. Fix the check for DATA_VERSION.<commit_after>/**
 * Copyright (C) 2009-2013 FoundationDB, LLC
 *
 * This program is free software: you can redistribute it and/or modify
 * it under the terms of the GNU Affero General Public License as published by
 * the Free Software Foundation, either version 3 of the License, or
 * (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU Affero General Public License for more details.
 *
 * You should have received a copy of the GNU Affero General Public License
 * along with this program.  If not, see <http://www.gnu.org/licenses/>.
 */

package com.foundationdb.server.store;

import com.foundationdb.ais.model.AkibanInformationSchema;
import com.foundationdb.ais.model.Columnar;
import com.foundationdb.ais.model.NameGenerator;
import com.foundationdb.ais.model.Routine;
import com.foundationdb.ais.model.SQLJJar;
import com.foundationdb.ais.model.Sequence;
import com.foundationdb.ais.model.SynchronizedNameGenerator;
import com.foundationdb.ais.model.Table;
import com.foundationdb.ais.model.TableName;
import com.foundationdb.ais.model.aisb2.AISBBasedBuilder;
import com.foundationdb.ais.model.aisb2.NewAISBuilder;
import com.foundationdb.ais.model.validation.AISValidations;
import com.foundationdb.ais.protobuf.ProtobufReader;
import com.foundationdb.ais.protobuf.ProtobufWriter;
import com.foundationdb.ais.protobuf.ProtobufWriter.WriteSelector;
import com.foundationdb.ais.util.UuidAssigner;
import com.foundationdb.qp.memoryadapter.BasicFactoryBase;
import com.foundationdb.qp.memoryadapter.MemoryAdapter;
import com.foundationdb.qp.memoryadapter.MemoryGroupCursor;
import com.foundationdb.qp.row.ValuesRow;
import com.foundationdb.qp.row.Row;
import com.foundationdb.qp.rowtype.RowType;
import com.foundationdb.server.PersistitAccumulatorTableStatusCache;
import com.foundationdb.server.error.AkibanInternalException;
import com.foundationdb.server.rowdata.RowDefCache;
import com.foundationdb.server.service.config.ConfigurationService;
import com.foundationdb.server.service.security.SecurityService;
import com.foundationdb.server.service.session.Session;
import com.foundationdb.server.service.session.SessionService;
import com.foundationdb.server.service.transaction.TransactionService;
import com.foundationdb.server.service.tree.TreeLink;
import com.foundationdb.server.service.tree.TreeService;
import com.foundationdb.server.service.tree.TreeVisitor;
import com.foundationdb.server.store.TableChanges.ChangeSet;
import com.foundationdb.server.store.format.PersistitStorageDescription;
import com.foundationdb.server.store.format.PersistitStorageFormatRegistry;
import com.foundationdb.server.util.ReadWriteMap;
import com.google.inject.Inject;
import com.persistit.Accumulator;
import com.persistit.Exchange;
import com.persistit.Key;
import com.persistit.KeyFilter;
import com.persistit.exception.PersistitException;
import com.persistit.exception.RollbackException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.nio.ByteBuffer;
import java.util.Collection;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Callable;
import java.util.concurrent.DelayQueue;
import java.util.concurrent.Delayed;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;

import static com.foundationdb.qp.storeadapter.PersistitAdapter.wrapPersistitException;
import static com.foundationdb.server.service.transaction.TransactionService.Callback;
import static com.foundationdb.server.service.transaction.TransactionService.CallbackType;
import static com.foundationdb.server.service.tree.TreeService.SCHEMA_TREE_NAME;

/**
 * Version 1.9.2 - 2013-10
 * <pre>
 *     - Data format version for non-SchemaManager (i.e. row/index) contents
 *     - Metadata format version for SchemaManager contents
 *     - Accumulator, at {@link #ACCUMULATOR_INDEX_SCHEMA_GEN}, for AIS generation
 *     - Serialized Protobuf per schema, via {@link ProtobufWriter}
 *     - Removed trees stored, with id and timestamp, for cleanup at next start
 *     - In-progress DDL, identified by session ID, stores two chunks of data:
 *         - Each affected table gets a key mapping to the DDL ID
 *         - New serialized Protobuf stored for each affected schema
 *
 *     {@link TreeService#SCHEMA_TREE_NAME}
 *         Accumulator[0]                           =>  Seq (generation)
 *         Accumulator[1]                           =>  Seq (online IDs)
 *         "metaVersion"                            =>  long
 *         "dataVersion"                            =>  long
 *         "delayed",(long)id,(long)ts              =>  "schema","tree"
 *         ...
 *         "protobuf","schema"                      =>  byte[] (AIS Protobuf)
 *         ...
 *         "protobufMem"                            =>  byte[] (AIS Protobuf)
 *         ...
 *         "online",(long)id                        =>  long (generation)
 *         ...
 *         "online",(long)id,"change",(int)tid      =>  byte[] (ChangeSet Protobuf)
 *         ...
 *         "online",(long)id,"protobuf","schema"    =>  byte[] (AIS Protobuf)
 *         ...
 * </pre>
 *
 * Version 1.2.1 - 2012-05
 * <pre>
 *     - Accumulator, at {@link #ACCUMULATOR_INDEX_SCHEMA_GEN}, for AIS generation
 *     - Serialized Protobuf per schema, via {@link ProtobufWriter}
 *     - Single "version" stored with every schema entry
 *     - Removed trees stored, with id and timestamp, for cleanup at next start
 *
 *     {@link TreeService#SCHEMA_TREE_NAME}
 *         Accumulator[0]                       =>  Seq
 *         "delayedTree",(long)id,(long)ts      =>  "schema","treeName"
 *         ...
 *         "by","PBAIS",(int)ver,"schema"       =>  byte[]
 *         ...
 *         "by","PBMEMAIS"                      =>  byte[]
 * </pre>
 *
 * Version 0.4 - 2011-05
 * <pre>
 *     - Single k/v pair containing the entire serialized AIS, via the
 *       deleted com.foundationdb.ais.metamodel.io.MessageTarget.
 *
 *     {@link TreeService#SCHEMA_TREE_NAME}
 *         "byAIS"      =>  byte[]
 * </pre>
 */
public class PersistitStoreSchemaManager extends AbstractSchemaManager {
    private static enum GenValue { NEW, SNAPSHOT }
    private static enum GenMap { PUT_NEW, NO_PUT }

    private static class SharedAIS {
        public final AtomicInteger refCount;
        public final AkibanInformationSchema ais;

        public SharedAIS(AkibanInformationSchema ais) {
            this.refCount = new AtomicInteger(0);
            this.ais = ais;
        }

        public int acquire() {
            return refCount.incrementAndGet();
        }

        public int release() {
            int count = refCount.decrementAndGet();
            assert count >= 0 : count;
            return count;
        }

        public int shareCount() {
            return refCount.get();
        }

        public boolean isShared() {
            return refCount.get() > 0;
        }

        @Override
        public String toString() {
            return "Shared[" + refCount.get() + "]" + ais;
        }
    }

    private static class AISAndTimestamp {
        public final SharedAIS sAIS;
        public final long timestamp;

        public AISAndTimestamp(SharedAIS sAIS, long timestamp) {
            this.sAIS = sAIS;
            this.timestamp = timestamp;
        }

        public boolean isUsableForStartTime(long startTime) {
            return startTime > timestamp;
        }

        @Override
        public String toString() {
            return sAIS + ":" + timestamp;
        }
    }

    private static final long CURRENT_DATA_VERSION = 2;
    private static final long CURRENT_META_VERSION = 1;
    private static final String S_K_META_VERSION = "metaVersion";
    private static final String S_K_DATA_VERSION = "dataVersion";
    private static final String S_K_DELAYED = "delayed";
    private static final String S_K_PROTOBUF = "protobuf";
    private static final String S_K_PROTOBUF_MEM = "protobufMem";
    private static final String S_K_ONLINE = "online";
    private static final String S_K_CHANGE = "change";
    private static final int ACCUMULATOR_INDEX_SCHEMA_GEN = 0;
    private static final int ACCUMULATOR_INDEX_ONLINE_ID = 1;

    /** Used when a consistent volume is required (e.g. accumulator) no matter what. */
    private static final String SCHEMA_TREE_INTERNAL_SCHEMA = "pssm";
    private static final Session.Key<SharedAIS> SESSION_SAIS_KEY = Session.Key.named("PSSM_SAIS");

    /**
     * <p>This is used as unusable cache identifier, count for outstanding DDLs, and sync object for updating
     * the latest cache value.</p>
     * <p>Why are all needed? Consider the simple case two concurrent DDLs, A and B. Both are have pre-committed
     * and cleared the cache. If A commits, the cache is updated, and then B commits, any reader using the cache
     * based on timestamp alone will get a stale snapshot. So, the cache can *only* be updated when there are no
     * more outstanding.</p>
     */
    private static final AISAndTimestamp CACHE_SENTINEL = new AISAndTimestamp(new SharedAIS(null), Long.MAX_VALUE);

    private static final Logger LOG = LoggerFactory.getLogger(PersistitStoreSchemaManager.class);

    private final TreeService treeService;
    private RowDefCache rowDefCache;
    private NameGenerator nameGenerator;
    private AtomicLong delayedTreeIDGenerator;
    private ReadWriteMap<Long,SharedAIS> aisMap;
    private volatile AISAndTimestamp latestAISCache;
    private TransactionService.Callback clearLatestCacheCallback;
    private TransactionService.Callback enqueueClearAndUpdateCallback;
    private BlockingQueue<QueueTask> taskQueue;
    private Thread queueConsumer;
    private AtomicInteger aisCacheMissCount;
    private AtomicInteger loadAISFromStorageCount;
    private AtomicInteger delayedTreeCount;

    @Inject
    public PersistitStoreSchemaManager(ConfigurationService config, SessionService sessionService,
                                       TreeService treeService, TransactionService txnService) {
        super(config, sessionService, txnService, new PersistitStorageFormatRegistry());
        this.treeService = treeService;
    }

    @Override
    protected NameGenerator getNameGenerator(Session session) {
        return nameGenerator;
    }

    @Override
    public void setSecurityService(SecurityService securityService) {
        this.securityService = securityService;
    }

    @Override
    public void addOnlineChangeSet(Session session, ChangeSet changeSet) {
        OnlineSession onlineSession = getOnlineSession(session, true);
        Exchange ex = getInternalExchange(session);
        try {
            ex.clear().append(S_K_ONLINE).append(onlineSession.id).append(S_K_CHANGE).append(changeSet.getTableId());
            ex.getValue().putByteArray(ChangeSetHelper.save(changeSet));
            ex.store();
        } catch(PersistitException | RollbackException e) {
            if(ex != null) {
                treeService.releaseExchange(session, ex);
            }
        }
    }

    @Override
    public AkibanInformationSchema getAis(Session session) {
        AkibanInformationSchema ais = getAISInternal(session).ais;
        OnlineSession onlineSession = getOnlineSession(session, null);
        if(onlineSession != null) {
            AkibanInformationSchema onlineAIS = getOnlineCache(session, ais).onlineToAIS.get(onlineSession.id);
            if(onlineAIS != null) {
                ais = onlineAIS;
            }
        }
        return ais;
    }

    private SharedAIS getAISInternal(Session session) {
        SharedAIS local = session.get(SESSION_SAIS_KEY);
        if(local != null) {
            return local;
        }

        // Latest is a volatile read and that is all that is required. If it is a valid cache, it can only
        // transition to non-valid (pre-committed DDL that doesn't affect out read validity) or the same
        // snapshot with a higher timestamp. See checks in updateLatestAISCache.
        final AISAndTimestamp cached = latestAISCache;
        final long startTimestamp = txnService.getTransactionStartTimestamp(session);
        if(cached.isUsableForStartTime(startTimestamp)) {
            local = cached.sAIS;
        }

        // Couldn't use the cache, do an always accurate accumulator lookup and check in map.
        long generation = 0;
        if(local == null) {
            aisCacheMissCount.incrementAndGet();
            generation = getGenerationSnapshot(session);
            local = aisMap.get(generation);
        }

        // Wasn't in map so need to reload from disk.
        // Should be 1) very rare and 2) fairly quick, so just do it under write lock to avoid duplicate work/entries
        if(local == null) {
            aisMap.claimExclusive();
            try {
                // Double check while while under exclusive
                local = aisMap.get(generation);
                if(local == null) {
                    loadAISFromStorageCount.incrementAndGet();
                    try {
                        local = loadToShared(session, GenValue.SNAPSHOT, GenMap.PUT_NEW);
                        buildRowDefCache(session, local.ais);
                    } catch(PersistitException | RollbackException e) {
                        throw wrapPersistitException(session, e);
                    }
                }
            } finally {
                aisMap.releaseExclusive();
            }
        }

        attachToSession(session, local);
        return local;
    }

    void treeWasRemoved(Session session, String schema, String treeName) {
        LOG.debug("Delaying removal of tree (until next restart): {}", treeName);
        delayedTreeCount.incrementAndGet();
        Exchange ex = schemaTreeExchange(session, schema);
        try {
            long id = delayedTreeIDGenerator.incrementAndGet();
            long timestamp = System.currentTimeMillis();
            ex.clear().append(S_K_DELAYED).append(id).append(timestamp);
            ex.getValue().setStreamMode(true);
            ex.getValue().put(schema);
            ex.getValue().put(treeName);
            ex.getValue().setStreamMode(false);
            ex.store();
        } catch(PersistitException | RollbackException e) {
            throw wrapPersistitException(session, e);
        } finally {
            treeService.releaseExchange(session, ex);
        }
    }

    @Override
    public void start() {
        super.start();

        rowDefCache = new RowDefCache(treeService.getTableStatusCache());
        boolean skipAISUpgrade = Boolean.parseBoolean(config.getProperty(SKIP_AIS_UPGRADE_PROPERTY));

        this.aisMap = ReadWriteMap.wrapNonFair(new HashMap<Long,SharedAIS>());

        final int[] ordinalChangeCount = { 0 };
        final AkibanInformationSchema newAIS;
        try(Session session = sessionService.createSession()) {
            newAIS = txnService.run(session, new Callable<AkibanInformationSchema>() {
                @Override
                public AkibanInformationSchema call() {
                    try {
                        // Unrelated to loading, but fine time to do it
                        cleanupDelayedTrees(session, true);

                        SharedAIS sAIS = loadToShared(session, GenValue.SNAPSHOT, GenMap.PUT_NEW);

                        // Migrate requires RowDefs, but shouldn't generate ordinals (otherwise nulls will be lost)
                        buildRowDefCache(session, sAIS.ais, true);
                        ordinalChangeCount[0] = migrateAccumulatorOrdinals(sAIS.ais);
                        // And generate it fully now that ordinals are definitely set
                        buildRowDefCache(session, sAIS.ais, false);

                        long startTimestamp = txnService.getTransactionStartTimestamp(session);
                        sAIS.acquire(); // So count while in cache is 1
                        latestAISCache = new AISAndTimestamp(sAIS, startTimestamp);
                        return sAIS.ais;
                    } catch(PersistitException e) {
                        throw wrapPersistitException(session, e);
                    }
                }
            });
        }

        this.nameGenerator = SynchronizedNameGenerator.wrap(new PersistitNameGenerator(newAIS));
        this.delayedTreeIDGenerator = new AtomicLong();
        for(Table table : newAIS.getTables().values()) {
            // Note: table.getVersion may be null (pre-1.4.3 volumes)
            tableVersionMap.put(table.getTableId(), table.getVersion());
        }

        this.taskQueue = new DelayQueue<>();
        this.queueConsumer = new Thread(new QueueConsumer(this.taskQueue), "PSSM_QUEUE");
        this.queueConsumer.start();

        this.clearLatestCacheCallback = new Callback() {
            @Override
            public void run(Session session, long timestamp) {
                updateLatestAISCache(CACHE_SENTINEL);
            }
        };
        this.enqueueClearAndUpdateCallback = new Callback() {
            @Override
            public void run(Session session, long timestamp) {
                taskQueue.add(new UpdateLatestCacheTask(0));
                taskQueue.add(new ClearAISMapTask(0, 10000));
            }
        };

        this.aisCacheMissCount = new AtomicInteger(0);
        this.loadAISFromStorageCount = new AtomicInteger(0);
        this.delayedTreeCount = new AtomicInteger(0);

        if (!skipAISUpgrade) {
            final AkibanInformationSchema upgradeAIS = aisCloner.clone(newAIS);
            UuidAssigner uuidAssigner = new UuidAssigner();
            upgradeAIS.visit(uuidAssigner);
            if(uuidAssigner.assignedAny() || ordinalChangeCount[0] > 0) {
                try(Session session = sessionService.createSession()) {
                    txnService.run(session, new Runnable() {
                        @Override
                        public void run() {
                            storedAISChange(session, upgradeAIS, upgradeAIS.getSchemas().keySet());
                        }
                    });
                }
            }
        }
        // else LOG.warn("Skipping AIS upgrade");

        registerSummaryTable();
    }

    @Override
    public void stop() {
        stopConsumer();
        this.delayedTreeIDGenerator = null;
        this.aisMap = null;
        this.nameGenerator = null;
        this.latestAISCache = null;
        this.clearLatestCacheCallback = null;
        this.enqueueClearAndUpdateCallback = null;
        this.taskQueue.clear();
        this.taskQueue = null;
        this.queueConsumer = null;
        this.aisCacheMissCount = null;
        this.loadAISFromStorageCount = null;
        this.delayedTreeCount = null;
        CACHE_SENTINEL.sAIS.refCount.set(0);
        super.stop();
    }

    @Override
    public void crash() {
        stop();
    }

    private void stopConsumer() {
        final long endNanoTime = System.nanoTime() + TimeUnit.SECONDS.toNanos(5);
        if(!taskQueue.offer(new StopTask(0))) {
            // DelayedQueue is unbounded, should never happen
            LOG.error("Could not offer StopTask");
        }
        for(;;) {
            long remaining = endNanoTime - System.nanoTime();
            if(remaining <= 0 || !queueConsumer.isAlive()) {
                break;
            }
            try {
                queueConsumer.join(100);
            } catch(InterruptedException e) {
                LOG.warn("Interrupted while trying to stop QueueConsumer");
            }
        }
    }

    private SharedAIS loadToShared(Session session, GenValue genValue, GenMap genMap) throws PersistitException {
        ProtobufReader reader = new ProtobufReader(storageFormatRegistry, new AkibanInformationSchema());
        loadFromStorage(session, reader);
        AkibanInformationSchema newAIS = finishReader(reader);
        return createValidatedShared(session, newAIS, genValue, genMap);
    }

    private void loadFromStorage(Session session, final ProtobufReader reader) throws PersistitException {
        treeService.visitStorage(session, new TreeVisitor() {
            @Override
            public void visit(Exchange ex) throws PersistitException {
                loadPrimaryProtobuf(ex, reader, null);
            }},
            SCHEMA_TREE_NAME
        );
    }

    private void loadProtobufChildren(Exchange ex, ProtobufReader reader, Collection<String> skipSchemas) throws PersistitException {
        ex.append(Key.BEFORE);
        while(ex.next()) {
            if((skipSchemas != null) && skipSchemas.contains(ex.getKey().indexTo(-1).decodeString())) {
                continue;
            }
            byte[] storedAIS = ex.getValue().getByteArray();
            ByteBuffer buffer = ByteBuffer.wrap(storedAIS);
            reader.loadBuffer(buffer);
        }
    }

    private void loadPrimaryProtobuf(Exchange ex, ProtobufReader reader, Collection<String> skipSchemas) throws PersistitException {
        ex.clear().append(S_K_META_VERSION).fetch();
        if(!ex.getValue().isDefined()) {
            // Can only be empty if there is no data here
            ex.clear();
            if(ex.hasNext()) {
                throw new AkibanInternalException("Unsupported data volume (no metadata version key)");
            } else {
                // No data in this volume at all.
                return;
            }
        }
        long metaVersion = ex.getValue().getLong();
        if(CURRENT_META_VERSION != metaVersion) {
            throw new AkibanInternalException(String.format("Unsupported data volume meta version %d, current %d",
                                                            metaVersion, CURRENT_META_VERSION));
        }

        ex.clear().append(S_K_DATA_VERSION).fetch();
        assert ex.getValue().isDefined() : "No data version";
        long dataVersion = ex.getValue().getLong();
        if(CURRENT_DATA_VERSION != dataVersion) {
            throw new AkibanInternalException(String.format("Unsupported data volume data version %d, current %d",
                                                            dataVersion, CURRENT_DATA_VERSION));
        }

        ex.clear().append(S_K_PROTOBUF);
        loadProtobufChildren(ex, reader, skipSchemas);

        ex.clear().append(S_K_PROTOBUF_MEM).fetch();
        if(ex.getValue().isDefined()) {
            byte[] storedAIS = ex.getValue().getByteArray();
            ByteBuffer buffer = ByteBuffer.wrap(storedAIS);
            reader.loadBuffer(buffer);
        }
    }

{'fraction_non_alphanumeric': 0.0908276720767164, 'fraction_numerical': 0.0034076760086502547, 'mean_word_length': 2.4184587813620073, 'pattern_counts': {'":': 2, '<': 43, '<?xml version=': 0, '>': 79, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 14, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2499919', 'n_tokens_mistral': 12553, 'n_tokens_neox': 12092, 'n_words': 2929}