BADStatementExecutor.java

/*
 * 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.
 */
package org.apache.asterix.bad.lang;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutorService;

import org.apache.asterix.app.active.ActiveNotificationHandler;
import org.apache.asterix.app.translator.QueryTranslator;
import org.apache.asterix.app.translator.RequestParameters;
import org.apache.asterix.bad.BADJobService;
import org.apache.asterix.bad.lang.statement.BrokerDropStatement;
import org.apache.asterix.bad.lang.statement.ChannelDropStatement;
import org.apache.asterix.bad.lang.statement.ProcedureDropStatement;
import org.apache.asterix.bad.metadata.Broker;
import org.apache.asterix.bad.metadata.Channel;
import org.apache.asterix.bad.metadata.DeployedJobSpecEventListener;
import org.apache.asterix.bad.metadata.Procedure;
import org.apache.asterix.common.dataflow.ICcApplicationContext;
import org.apache.asterix.common.exceptions.CompilationException;
import org.apache.asterix.common.functions.FunctionSignature;
import org.apache.asterix.compiler.provider.ILangCompilationProvider;
import org.apache.asterix.lang.common.base.Statement;
import org.apache.asterix.lang.common.statement.CreateIndexStatement;
import org.apache.asterix.lang.common.statement.DataverseDropStatement;
import org.apache.asterix.lang.common.statement.DropDatasetStatement;
import org.apache.asterix.lang.common.statement.FunctionDropStatement;
import org.apache.asterix.lang.common.statement.IndexDropStatement;
import org.apache.asterix.lang.common.struct.Identifier;
import org.apache.asterix.metadata.MetadataManager;
import org.apache.asterix.metadata.MetadataTransactionContext;
import org.apache.asterix.metadata.declared.MetadataProvider;
import org.apache.asterix.metadata.entities.Dataverse;
import org.apache.asterix.metadata.entities.Function;
import org.apache.asterix.translator.IRequestParameters;
import org.apache.asterix.translator.SessionOutput;
import org.apache.hyracks.algebricks.common.exceptions.AlgebricksException;
import org.apache.hyracks.algebricks.common.utils.Pair;
import org.apache.hyracks.api.client.IHyracksClientConnection;

public class BADStatementExecutor extends QueryTranslator {

    public BADStatementExecutor(ICcApplicationContext appCtx, List<Statement> statements, SessionOutput output,
            ILangCompilationProvider compliationProvider, ExecutorService executorService) {
        super(appCtx, statements, output, compliationProvider, executorService);
    }

    //TODO: Most of this file could go away if we had metadata dependencies

    private Pair<List<Channel>, List<Procedure>> checkIfDatasetIsInUse(MetadataTransactionContext mdTxnCtx,
            String dataverse, String dataset, boolean checkAll) throws AlgebricksException {
        List<Channel> channelsUsingDataset = new ArrayList<>();
        List<Procedure> proceduresUsingDataset = new ArrayList<>();
        List<Channel> channels = BADLangExtension.getAllChannels(mdTxnCtx);
        for (Channel channel : channels) {
            List<List<List<String>>> dependencies = channel.getDependencies();
            List<List<String>> datasetDependencies = dependencies.get(0);
            for (List<String> dependency : datasetDependencies) {
                if (dependency.get(0).equals(dataverse) && dependency.get(1).equals(dataset)) {
                    channelsUsingDataset.add(channel);
                    if (!checkAll) {
                        return new Pair<>(channelsUsingDataset, proceduresUsingDataset);
                    }

                }
            }

        }
        List<Procedure> procedures = BADLangExtension.getAllProcedures(mdTxnCtx);
        for (Procedure procedure : procedures) {
            List<List<List<String>>> dependencies = procedure.getDependencies();
            List<List<String>> datasetDependencies = dependencies.get(0);
            for (List<String> dependency : datasetDependencies) {
                if (dependency.get(0).equals(dataverse) && dependency.get(1).equals(dataset)) {
                    proceduresUsingDataset.add(procedure);
                    if (!checkAll) {
                        return new Pair<>(channelsUsingDataset, proceduresUsingDataset);
                    }
                }
            }

        }
        return new Pair<>(channelsUsingDataset, proceduresUsingDataset);
    }

    private Pair<List<Channel>, List<Procedure>> checkIfFunctionIsInUse(MetadataTransactionContext mdTxnCtx,
            String dvId, String function, String arity, boolean checkAll)
            throws CompilationException, AlgebricksException {
        List<Channel> channelsUsingFunction = new ArrayList<>();
        List<Procedure> proceduresUsingFunction = new ArrayList<>();

        List<Channel> channels = BADLangExtension.getAllChannels(mdTxnCtx);
        for (Channel channel : channels) {
            List<List<List<String>>> dependencies = channel.getDependencies();
            List<List<String>> datasetDependencies = dependencies.get(1);
            for (List<String> dependency : datasetDependencies) {
                if (dependency.get(0).equals(dvId) && dependency.get(1).equals(function)
                        && dependency.get(2).equals(arity)) {
                    channelsUsingFunction.add(channel);
                    if (!checkAll) {
                        return new Pair<>(channelsUsingFunction, proceduresUsingFunction);
                    }
                }
            }

        }
        List<Procedure> procedures = BADLangExtension.getAllProcedures(mdTxnCtx);
        for (Procedure procedure : procedures) {
            List<List<List<String>>> dependencies = procedure.getDependencies();
            List<List<String>> datasetDependencies = dependencies.get(1);
            for (List<String> dependency : datasetDependencies) {
                if (dependency.get(0).equals(dvId) && dependency.get(1).equals(function)
                        && dependency.get(2).equals(arity)) {
                    proceduresUsingFunction.add(procedure);
                    if (!checkAll) {
                        return new Pair<>(channelsUsingFunction, proceduresUsingFunction);
                    }
                }
            }

        }
        return new Pair<>(channelsUsingFunction, proceduresUsingFunction);
    }

    private void throwErrorIfDatasetUsed(MetadataTransactionContext mdTxnCtx, String dataverse, String dataset)
            throws CompilationException, AlgebricksException {
        Pair<List<Channel>, List<Procedure>> dependents = checkIfDatasetIsInUse(mdTxnCtx, dataverse, dataset, false);
        if (dependents.first.size() > 0) {
            throw new CompilationException("Cannot alter dataset " + dataverse + "." + dataset + ". "
                    + dependents.first.get(0).getChannelId() + " depends on it!");
        }
        if (dependents.second.size() > 0) {
            throw new CompilationException("Cannot alter dataset " + dataverse + "." + dataset + ". "
                    + dependents.second.get(0).getEntityId() + " depends on it!");
        }
    }

    private void throwErrorIfFunctionUsed(MetadataTransactionContext mdTxnCtx, String dataverse, String function,
            String arity, FunctionSignature sig) throws CompilationException, AlgebricksException {
        Pair<List<Channel>, List<Procedure>> dependents =
                checkIfFunctionIsInUse(mdTxnCtx, dataverse, function, arity, false);
        String errorStart = sig != null ? "Cannot drop function " + sig + "." : "Cannot drop index.";
        if (dependents.first.size() > 0) {
            throw new CompilationException(
                    errorStart + " " + dependents.first.get(0).getChannelId() + " depends on it!");
        }
        if (dependents.second.size() > 0) {
            throw new CompilationException(
                    errorStart + " " + dependents.second.get(0).getEntityId() + " depends on it!");
        }
    }

    @Override
    public void handleDatasetDropStatement(MetadataProvider metadataProvider, Statement stmt,
            IHyracksClientConnection hcc, IRequestParameters requestParameters) throws Exception {
        MetadataTransactionContext mdTxnCtx = MetadataManager.INSTANCE.beginTransaction();
        metadataProvider.setMetadataTxnContext(mdTxnCtx);
        String dvId = getActiveDataverse(((DropDatasetStatement) stmt).getDataverseName());
        Identifier dsId = ((DropDatasetStatement) stmt).getDatasetName();

        throwErrorIfDatasetUsed(mdTxnCtx, dvId, dsId.getValue());

        MetadataManager.INSTANCE.commitTransaction(mdTxnCtx);
        super.handleDatasetDropStatement(metadataProvider, stmt, hcc, requestParameters);
    }

    @Override
    public void handleCreateIndexStatement(MetadataProvider metadataProvider, Statement stmt,
            IHyracksClientConnection hcc, IRequestParameters requestParameters) throws Exception {

        MetadataTransactionContext mdTxnCtx = MetadataManager.INSTANCE.beginTransaction();
        metadataProvider.setMetadataTxnContext(mdTxnCtx);
        //Allow channels to use the new index
        String dvId = getActiveDataverse(((CreateIndexStatement) stmt).getDataverseName());
        String dsId = ((CreateIndexStatement) stmt).getDatasetName().getValue();

        Pair<List<Channel>, List<Procedure>> usages = checkIfDatasetIsInUse(mdTxnCtx, dvId, dsId, true);

        List<Dataverse> dataverseList = MetadataManager.INSTANCE.getDataverses(mdTxnCtx);
        for (Dataverse dv : dataverseList) {
            List<Function> functions = MetadataManager.INSTANCE.getFunctions(mdTxnCtx, dv.getDataverseName());
            for (Function function : functions) {
                for (List<String> datasetDependency : function.getDependencies().get(0)) {
                    if (datasetDependency.get(0).equals(dvId) && datasetDependency.get(1).equals(dsId)) {
                        Pair<List<Channel>, List<Procedure>> functionUsages =
                                checkIfFunctionIsInUse(mdTxnCtx, function.getDataverseName(), function.getName(),
                                        Integer.toString(function.getArity()), true);
                        for (Channel channel : functionUsages.first) {
                            if (!usages.first.contains(channel)) {
                                usages.first.add(channel);
                            }
                        }
                        for (Procedure procedure : functionUsages.second) {
                            if (!usages.second.contains(procedure)) {
                                usages.second.add(procedure);
                            }
                        }
                    }
                }
            }
        }

        ActiveNotificationHandler activeEventHandler =
                (ActiveNotificationHandler) appCtx.getActiveNotificationHandler();

        for (Channel channel : usages.first) {
            DeployedJobSpecEventListener listener =
                    (DeployedJobSpecEventListener) activeEventHandler.getListener(channel.getChannelId());
            listener.suspend();
        }
        for (Procedure procedure : usages.second) {
            DeployedJobSpecEventListener listener =
                    (DeployedJobSpecEventListener) activeEventHandler.getListener(procedure.getEntityId());
            listener.suspend();
        }

        MetadataManager.INSTANCE.commitTransaction(mdTxnCtx);
        metadataProvider.getLocks().unlock();

        metadataProvider = new MetadataProvider(appCtx, activeDataverse);
        super.handleCreateIndexStatement(metadataProvider, stmt, hcc, requestParameters);

        for (Channel channel : usages.first) {
            metadataProvider = new MetadataProvider(appCtx, activeDataverse);
            BADJobService.redeployJobSpec(channel.getChannelId(), channel.getChannelBody(), metadataProvider, this, hcc,
                    requestParameters, false);
            metadataProvider.getLocks().unlock();
        }
        for (Procedure procedure : usages.second) {
            metadataProvider = new MetadataProvider(appCtx, activeDataverse);
            BADJobService.redeployJobSpec(procedure.getEntityId(), procedure.getBody(), metadataProvider, this, hcc,
                    requestParameters, false);
            metadataProvider.getLocks().unlock();
        }

    }

    @Override
    protected void handleIndexDropStatement(MetadataProvider metadataProvider, Statement stmt,
            IHyracksClientConnection hcc, IRequestParameters requestParameters) throws Exception {
        MetadataTransactionContext mdTxnCtx = MetadataManager.INSTANCE.beginTransaction();
        metadataProvider.setMetadataTxnContext(mdTxnCtx);
        String dvId = getActiveDataverse(((IndexDropStatement) stmt).getDataverseName());
        Identifier dsId = ((IndexDropStatement) stmt).getDatasetName();

        throwErrorIfDatasetUsed(mdTxnCtx, dvId, dsId.getValue());

        List<Dataverse> dataverseList = MetadataManager.INSTANCE.getDataverses(mdTxnCtx);
        for (Dataverse dv : dataverseList) {
            List<Function> functions = MetadataManager.INSTANCE.getFunctions(mdTxnCtx, dv.getDataverseName());
            for (Function function : functions) {
                for (List<String> datasetDependency : function.getDependencies().get(0)) {
                    if (datasetDependency.get(0).equals(dvId) && datasetDependency.get(1).equals(dsId.getValue())) {
                        throwErrorIfFunctionUsed(mdTxnCtx, function.getDataverseName(), function.getName(),
                                Integer.toString(function.getArity()), null);
                    }
                }
            }
        }

        MetadataManager.INSTANCE.commitTransaction(mdTxnCtx);
        super.handleIndexDropStatement(metadataProvider, stmt, hcc, requestParameters);
    }

    @Override
    protected void handleFunctionDropStatement(MetadataProvider metadataProvider, Statement stmt) throws Exception {
        MetadataTransactionContext mdTxnCtx = MetadataManager.INSTANCE.beginTransaction();
        metadataProvider.setMetadataTxnContext(mdTxnCtx);
        FunctionSignature sig = ((FunctionDropStatement) stmt).getFunctionSignature();

        String dvId = getActiveDataverseName(sig.getNamespace());
        String function = sig.getName();
        String arity = Integer.toString(sig.getArity());

        throwErrorIfFunctionUsed(mdTxnCtx, dvId, function, arity, sig);

        MetadataManager.INSTANCE.commitTransaction(mdTxnCtx);
        super.handleFunctionDropStatement(metadataProvider, stmt);
    }

    @Override
    protected void handleDataverseDropStatement(MetadataProvider metadataProvider, Statement stmt,
            IHyracksClientConnection hcc) throws Exception {
        MetadataTransactionContext mdTxnCtx = MetadataManager.INSTANCE.beginTransaction();
        metadataProvider.setMetadataTxnContext(mdTxnCtx);
        Identifier dvId = ((DataverseDropStatement) stmt).getDataverseName();
        MetadataProvider tempMdProvider = new MetadataProvider(appCtx, metadataProvider.getDefaultDataverse());
        tempMdProvider.getConfig().putAll(metadataProvider.getConfig());
        List<Channel> channels = BADLangExtension.getAllChannels(mdTxnCtx);
        for (Channel channel : channels) {
            if (channel.getChannelId().getDataverse().equals(dvId.getValue())) {
                continue;
            }
            List<List<List<String>>> dependencies = channel.getDependencies();
            for (List<List<String>> dependencyList : dependencies) {
                for (List<String> dependency : dependencyList) {
                    if (dependency.get(0).equals(dvId.getValue())) {
                        throw new CompilationException("Cannot drop dataverse " + dvId.getValue() + ". "
                                + channel.getChannelId() + " depends on it!");
                    }
                }
            }
        }
        List<Procedure> procedures = BADLangExtension.getAllProcedures(mdTxnCtx);
        for (Procedure procedure : procedures) {
            if (procedure.getEntityId().getDataverse().equals(dvId.getValue())) {
                continue;
            }
            List<List<List<String>>> dependencies = procedure.getDependencies();
            for (List<List<String>> dependencyList : dependencies) {
                for (List<String> dependency : dependencyList) {
                    if (dependency.get(0).equals(dvId.getValue())) {
                        throw new CompilationException("Cannot drop dataverse " + dvId.getValue() + ". "
                                + procedure.getEntityId() + " depends on it!");
                    }
                }
            }
        }
        final IRequestParameters requestParameters =
                new RequestParameters(null, null, null, null, null, null, null, true);
        for (Channel channel : channels) {
            if (!channel.getChannelId().getDataverse().equals(dvId.getValue())) {
                continue;
            }
            tempMdProvider.getLocks().reset();
            ChannelDropStatement drop =
                    new ChannelDropStatement(dvId, new Identifier(channel.getChannelId().getEntityName()), false);
            drop.handle(hcc, this, requestParameters, tempMdProvider, 0, null);
        }
        for (Procedure procedure : procedures) {
            if (!procedure.getEntityId().getDataverse().equals(dvId.getValue())) {
                continue;
            }
            tempMdProvider.getLocks().reset();
            ProcedureDropStatement drop = new ProcedureDropStatement(new FunctionSignature(dvId.getValue(),
                    procedure.getEntityId().getEntityName(), procedure.getArity()), false);
            drop.handle(hcc, this, requestParameters, tempMdProvider, 0, null);
        }
        List<Broker> brokers = BADLangExtension.getBrokers(mdTxnCtx, dvId.getValue());
        for (Broker broker : brokers) {
            tempMdProvider.getLocks().reset();
            BrokerDropStatement drop = new BrokerDropStatement(dvId, new Identifier(broker.getBrokerName()), false);
            drop.handle(hcc, this, requestParameters, tempMdProvider, 0, null);
        }
        MetadataManager.INSTANCE.commitTransaction(mdTxnCtx);
        super.handleDataverseDropStatement(metadataProvider, stmt, hcc);
    }

}