/*
 * 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.metadata.entities;

import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.stream.IntStream;

import org.apache.asterix.common.api.IDatasetInfoProvider;
import org.apache.asterix.common.api.ILSMComponentIdGeneratorFactory;
import org.apache.asterix.common.config.DatasetConfig.DatasetType;
import org.apache.asterix.common.context.CorrelatedPrefixMergePolicyFactory;
import org.apache.asterix.common.context.DatasetInfoProvider;
import org.apache.asterix.common.context.DatasetLSMComponentIdGeneratorFactory;
import org.apache.asterix.common.context.IStorageComponentProvider;
import org.apache.asterix.common.dataflow.NoOpFrameOperationCallbackFactory;
import org.apache.asterix.common.exceptions.CompilationException;
import org.apache.asterix.common.exceptions.ErrorCode;
import org.apache.asterix.common.ioopcallbacks.AtomicLSMIndexIOOperationCallbackFactory;
import org.apache.asterix.common.ioopcallbacks.LSMIndexIOOperationCallbackFactory;
import org.apache.asterix.common.ioopcallbacks.LSMIndexPageWriteCallbackFactory;
import org.apache.asterix.common.metadata.DatasetFullyQualifiedName;
import org.apache.asterix.common.metadata.DataverseName;
import org.apache.asterix.common.metadata.IDataset;
import org.apache.asterix.common.transactions.IRecoveryManager.ResourceType;
import org.apache.asterix.common.utils.JobUtils;
import org.apache.asterix.common.utils.JobUtils.ProgressState;
import org.apache.asterix.common.utils.StoragePathUtil;
import org.apache.asterix.external.feed.management.FeedConnectionId;
import org.apache.asterix.formats.nontagged.BinaryHashFunctionFactoryProvider;
import org.apache.asterix.formats.nontagged.SerializerDeserializerProvider;
import org.apache.asterix.formats.nontagged.TypeTraitProvider;
import org.apache.asterix.metadata.IDatasetDetails;
import org.apache.asterix.metadata.MetadataCache;
import org.apache.asterix.metadata.MetadataManager;
import org.apache.asterix.metadata.MetadataTransactionContext;
import org.apache.asterix.metadata.api.IMetadataEntity;
import org.apache.asterix.metadata.dataset.DatasetFormatInfo;
import org.apache.asterix.metadata.declared.ArrayBTreeResourceFactoryProvider;
import org.apache.asterix.metadata.declared.BTreeResourceFactoryProvider;
import org.apache.asterix.metadata.declared.MetadataProvider;
import org.apache.asterix.metadata.utils.DatasetUtil;
import org.apache.asterix.metadata.utils.IndexUtil;
import org.apache.asterix.metadata.utils.InvertedIndexResourceFactoryProvider;
import org.apache.asterix.metadata.utils.MetadataUtil;
import org.apache.asterix.metadata.utils.RTreeResourceFactoryProvider;
import org.apache.asterix.metadata.utils.TypeUtil;
import org.apache.asterix.om.functions.BuiltinFunctions;
import org.apache.asterix.om.types.ARecordType;
import org.apache.asterix.om.types.IAType;
import org.apache.asterix.runtime.compression.CompressionManager;
import org.apache.asterix.transaction.management.opcallbacks.AbstractIndexModificationOperationCallback.Operation;
import org.apache.asterix.transaction.management.opcallbacks.LockThenSearchOperationCallbackFactory;
import org.apache.asterix.transaction.management.opcallbacks.PrimaryIndexInstantSearchOperationCallbackFactory;
import org.apache.asterix.transaction.management.opcallbacks.PrimaryIndexModificationOperationCallbackFactory;
import org.apache.asterix.transaction.management.opcallbacks.PrimaryIndexOperationTrackerFactory;
import org.apache.asterix.transaction.management.opcallbacks.SecondaryIndexInstanctSearchOperationCallbackFactory;
import org.apache.asterix.transaction.management.opcallbacks.SecondaryIndexModificationOperationCallbackFactory;
import org.apache.asterix.transaction.management.opcallbacks.SecondaryIndexOperationTrackerFactory;
import org.apache.asterix.transaction.management.opcallbacks.SecondaryIndexSearchOperationCallbackFactory;
import org.apache.asterix.transaction.management.opcallbacks.UpsertOperationCallbackFactory;
import org.apache.asterix.transaction.management.resource.DatasetLocalResourceFactory;
import org.apache.asterix.transaction.management.runtime.CommitRuntimeFactory;
import org.apache.asterix.transaction.management.runtime.NoOpCommitRuntimeFactory;
import org.apache.asterix.transaction.management.service.transaction.DatasetIdFactory;
import org.apache.commons.lang3.mutable.MutableBoolean;
import org.apache.commons.lang3.mutable.MutableObject;
import org.apache.hyracks.algebricks.common.exceptions.AlgebricksException;
import org.apache.hyracks.algebricks.common.utils.Pair;
import org.apache.hyracks.algebricks.core.algebra.base.ILogicalOperator;
import org.apache.hyracks.algebricks.core.algebra.base.LogicalExpressionTag;
import org.apache.hyracks.algebricks.core.algebra.base.LogicalOperatorTag;
import org.apache.hyracks.algebricks.core.algebra.expressions.AbstractFunctionCallExpression;
import org.apache.hyracks.algebricks.core.algebra.operators.logical.UnnestOperator;
import org.apache.hyracks.algebricks.data.IBinaryComparatorFactoryProvider;
import org.apache.hyracks.algebricks.data.ISerializerDeserializerProvider;
import org.apache.hyracks.algebricks.data.ITypeTraitProvider;
import org.apache.hyracks.algebricks.runtime.base.IPushRuntimeFactory;
import org.apache.hyracks.api.client.IHyracksClientConnection;
import org.apache.hyracks.api.dataflow.value.IBinaryComparatorFactory;
import org.apache.hyracks.api.dataflow.value.IBinaryHashFunctionFactory;
import org.apache.hyracks.api.dataflow.value.ISerializerDeserializer;
import org.apache.hyracks.api.dataflow.value.ITuplePartitionerFactory;
import org.apache.hyracks.api.dataflow.value.ITypeTraits;
import org.apache.hyracks.api.dataflow.value.RecordDescriptor;
import org.apache.hyracks.api.exceptions.SourceLocation;
import org.apache.hyracks.api.io.FileSplit;
import org.apache.hyracks.api.job.JobSpecification;
import org.apache.hyracks.dataflow.common.data.partition.FieldHashPartitionerFactory;
import org.apache.hyracks.storage.am.common.api.IModificationOperationCallbackFactory;
import org.apache.hyracks.storage.am.common.api.ISearchOperationCallbackFactory;
import org.apache.hyracks.storage.am.common.dataflow.IndexDropOperatorDescriptor.DropOption;
import org.apache.hyracks.storage.am.common.impls.NoOpOperationCallbackFactory;
import org.apache.hyracks.storage.am.common.ophelpers.IndexOperation;
import org.apache.hyracks.storage.am.lsm.common.api.IFrameOperationCallbackFactory;
import org.apache.hyracks.storage.am.lsm.common.api.ILSMIOOperationCallbackFactory;
import org.apache.hyracks.storage.am.lsm.common.api.ILSMMergePolicyFactory;
import org.apache.hyracks.storage.am.lsm.common.api.ILSMOperationTrackerFactory;
import org.apache.hyracks.storage.am.lsm.common.api.ILSMPageWriteCallbackFactory;
import org.apache.hyracks.storage.common.IResourceFactory;
import org.apache.logging.log4j.Level;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;

/**
 * Metadata describing a dataset.
 */
public class Dataset implements IMetadataEntity<Dataset>, IDataset {

    /*
     * Constants
     */
    private static final long serialVersionUID = 2L;
    private static final Logger LOGGER = LogManager.getLogger();
    private static final RTreeResourceFactoryProvider rTreeResourceFactoryProvider =
            RTreeResourceFactoryProvider.INSTANCE;
    private static final InvertedIndexResourceFactoryProvider invertedIndexResourceFactoryProvider =
            InvertedIndexResourceFactoryProvider.INSTANCE;
    /*
     * Members
     */
    private final int datasetId;
    private final DataverseName dataverseName;
    private final String datasetName;
    private final DataverseName recordTypeDataverseName;
    private final String recordTypeName;
    private final String nodeGroupName;
    private final String compactionPolicyFactory;
    private final Map<String, String> hints;
    private final Map<String, String> compactionPolicyProperties;
    private final DatasetType datasetType;
    private final IDatasetDetails datasetDetails;
    private final DataverseName metaTypeDataverseName;
    private final String metaTypeName;
    private final long rebalanceCount;
    private int pendingOp;
    private final String compressionScheme;
    private final DatasetFullyQualifiedName datasetFullyQualifiedName;
    private final DatasetFormatInfo datasetFormatInfo;

    public Dataset(DataverseName dataverseName, String datasetName, DataverseName recordTypeDataverseName,
            String recordTypeName, String nodeGroupName, String compactionPolicy,
            Map<String, String> compactionPolicyProperties, IDatasetDetails datasetDetails, Map<String, String> hints,
            DatasetType datasetType, int datasetId, int pendingOp) {
        this(dataverseName, datasetName, recordTypeDataverseName, recordTypeName, /*metaTypeDataverseName*/null,
                /*metaTypeName*/null, nodeGroupName, compactionPolicy, compactionPolicyProperties, datasetDetails,
                hints, datasetType, datasetId, pendingOp, CompressionManager.NONE, DatasetFormatInfo.SYSTEM_DEFAULT);
    }

    public Dataset(DataverseName dataverseName, String datasetName, DataverseName itemTypeDataverseName,
            String itemTypeName, DataverseName metaItemTypeDataverseName, String metaItemTypeName, String nodeGroupName,
            String compactionPolicy, Map<String, String> compactionPolicyProperties, IDatasetDetails datasetDetails,
            Map<String, String> hints, DatasetType datasetType, int datasetId, int pendingOp, String compressionScheme,
            DatasetFormatInfo datasetFormatInfo) {
        this(dataverseName, datasetName, itemTypeDataverseName, itemTypeName, metaItemTypeDataverseName,
                metaItemTypeName, nodeGroupName, compactionPolicy, compactionPolicyProperties, datasetDetails, hints,
                datasetType, datasetId, pendingOp, 0L, compressionScheme, datasetFormatInfo);
    }

    public Dataset(Dataset dataset) {
        this(dataset.dataverseName, dataset.datasetName, dataset.recordTypeDataverseName, dataset.recordTypeName,
                dataset.metaTypeDataverseName, dataset.metaTypeName, dataset.nodeGroupName,
                dataset.compactionPolicyFactory, dataset.compactionPolicyProperties, dataset.datasetDetails,
                dataset.hints, dataset.datasetType, dataset.datasetId, dataset.pendingOp, dataset.rebalanceCount,
                dataset.compressionScheme, dataset.datasetFormatInfo);
    }

    public Dataset(DataverseName dataverseName, String datasetName, DataverseName itemTypeDataverseName,
            String itemTypeName, DataverseName metaItemTypeDataverseName, String metaItemTypeName, String nodeGroupName,
            String compactionPolicy, Map<String, String> compactionPolicyProperties, IDatasetDetails datasetDetails,
            Map<String, String> hints, DatasetType datasetType, int datasetId, int pendingOp, long rebalanceCount,
            String compressionScheme, DatasetFormatInfo datasetFormatInfo) {
        this.dataverseName = dataverseName;
        this.datasetName = datasetName;
        this.recordTypeName = itemTypeName;
        this.recordTypeDataverseName = itemTypeDataverseName;
        this.metaTypeDataverseName = metaItemTypeDataverseName;
        this.metaTypeName = metaItemTypeName;
        this.nodeGroupName = nodeGroupName;
        this.compactionPolicyFactory = compactionPolicy;
        this.compactionPolicyProperties = compactionPolicyProperties;
        this.datasetType = datasetType;
        this.datasetDetails = datasetDetails;
        this.datasetId = datasetId;
        this.pendingOp = pendingOp;
        this.hints = hints;
        this.rebalanceCount = rebalanceCount;
        this.compressionScheme = compressionScheme;
        datasetFullyQualifiedName = new DatasetFullyQualifiedName(dataverseName, datasetName);
        this.datasetFormatInfo = datasetFormatInfo;
    }

    @Override
    public DataverseName getDataverseName() {
        return dataverseName;
    }

    @Override
    public String getDatasetName() {
        return datasetName;
    }

    public String getItemTypeName() {
        return recordTypeName;
    }

    public DataverseName getItemTypeDataverseName() {
        return recordTypeDataverseName;
    }

    public String getNodeGroupName() {
        return nodeGroupName;
    }

    public String getCompactionPolicy() {
        return compactionPolicyFactory;
    }

    public Map<String, String> getCompactionPolicyProperties() {
        return compactionPolicyProperties;
    }

    public DatasetType getDatasetType() {
        return datasetType;
    }

    public IDatasetDetails getDatasetDetails() {
        return datasetDetails;
    }

    public Map<String, String> getHints() {
        return hints;
    }

    public int getDatasetId() {
        return datasetId;
    }

    public int getPendingOp() {
        return pendingOp;
    }

    public DataverseName getMetaItemTypeDataverseName() {
        return metaTypeDataverseName;
    }

    public String getMetaItemTypeName() {
        return metaTypeName;
    }

    public long getRebalanceCount() {
        return rebalanceCount;
    }

    public boolean hasMetaPart() {
        return metaTypeDataverseName != null && metaTypeName != null;
    }

    public void setPendingOp(int pendingOp) {
        this.pendingOp = pendingOp;
    }

    @Override
    public Dataset addToCache(MetadataCache cache) {
        return cache.addDatasetIfNotExists(this);
    }

    @Override
    public Dataset dropFromCache(MetadataCache cache) {
        return cache.dropDataset(this);
    }

    @Override
    public boolean equals(Object other) {
        if (this == other) {
            return true;
        }
        if (!(other instanceof Dataset)) {
            return false;
        }
        Dataset otherDataset = (Dataset) other;
        return Objects.equals(dataverseName, otherDataset.dataverseName)
                && Objects.equals(datasetName, otherDataset.datasetName);
    }

    public boolean allow(ILogicalOperator topOp, byte operation) {
        if (!hasMetaPart()) {
            return true;
        }
        if (topOp.getInputs().get(0).getValue().getOperatorTag() != LogicalOperatorTag.ASSIGN) {
            return false;
        }
        ILogicalOperator op = topOp.getInputs().get(0).getValue();
        while ((!op.getInputs().isEmpty())
                && op.getInputs().get(0).getValue().getOperatorTag() != LogicalOperatorTag.UNNEST) {
            op = op.getInputs().get(0).getValue();
        }
        if (op.getInputs().isEmpty()) {
            return false;
        }
        UnnestOperator unnestOp = (UnnestOperator) op.getInputs().get(0).getValue();
        if (unnestOp.getExpressionRef().getValue().getExpressionTag() != LogicalExpressionTag.FUNCTION_CALL) {
            return false;
        }
        AbstractFunctionCallExpression functionCall =
                (AbstractFunctionCallExpression) unnestOp.getExpressionRef().getValue();
        if (functionCall.getFunctionIdentifier() != BuiltinFunctions.FEED_COLLECT) {
            return false;
        }
        return operation == DatasetUtil.OP_UPSERT;
    }

    /**
     * Drop this dataset
     *
     * @param metadataProvider metadata provider that can be used to get metadata info and runtimes
     * @param mdTxnCtx         the transaction context
     * @param jobsToExecute    a list of jobs to be executed as part of the drop operation
     * @param bActiveTxn       whether the metadata transaction is ongoing
     * @param progress         a mutable progress state used for error handling during the drop operation
     * @param hcc              a client connection to hyracks master for job execution
     * @param sourceLoc
     * @throws Exception if an error occur during the drop process or if the dataset can't be dropped for any reason
     */
    public void drop(MetadataProvider metadataProvider, MutableObject<MetadataTransactionContext> mdTxnCtx,
            List<JobSpecification> jobsToExecute, MutableBoolean bActiveTxn, MutableObject<ProgressState> progress,
            IHyracksClientConnection hcc, boolean dropCorrespondingNodeGroup, SourceLocation sourceLoc,
            Set<DropOption> options, boolean force) throws Exception {
        Map<FeedConnectionId, Pair<JobSpecification, Boolean>> disconnectJobList = new HashMap<>();
        if (getDatasetType() == DatasetType.INTERNAL) {
            // #. prepare jobs to drop the datatset and the indexes in NC
            List<Index> indexes =
                    MetadataManager.INSTANCE.getDatasetIndexes(mdTxnCtx.getValue(), dataverseName, datasetName);
            for (int j = 0; j < indexes.size(); j++) {
                if (indexes.get(j).isSecondaryIndex()) {
                    jobsToExecute.add(IndexUtil.buildDropIndexJobSpec(indexes.get(j), metadataProvider, this, options,
                            sourceLoc));
                }
            }
            jobsToExecute.add(DatasetUtil.dropDatasetJobSpec(this, metadataProvider, options));
            // #. mark the existing dataset as PendingDropOp
            MetadataManager.INSTANCE.dropDataset(mdTxnCtx.getValue(), dataverseName, datasetName, force);
            MetadataManager.INSTANCE.addDataset(mdTxnCtx.getValue(),
                    new Dataset(dataverseName, datasetName, getItemTypeDataverseName(), getItemTypeName(),
                            getMetaItemTypeDataverseName(), getMetaItemTypeName(), getNodeGroupName(),
                            getCompactionPolicy(), getCompactionPolicyProperties(), getDatasetDetails(), getHints(),
                            getDatasetType(), getDatasetId(), MetadataUtil.PENDING_DROP_OP, getCompressionScheme(),
                            getDatasetFormatInfo()));

            MetadataManager.INSTANCE.commitTransaction(mdTxnCtx.getValue());
            bActiveTxn.setValue(false);
            progress.setValue(ProgressState.ADDED_PENDINGOP_RECORD_TO_METADATA);

            // # disconnect the feeds
            for (Pair<JobSpecification, Boolean> p : disconnectJobList.values()) {
                JobUtils.runJob(hcc, p.first, true);
            }

            // #. run the jobs
            for (JobSpecification jobSpec : jobsToExecute) {
                JobUtils.runJob(hcc, jobSpec, true);
            }

            mdTxnCtx.setValue(MetadataManager.INSTANCE.beginTransaction());
            bActiveTxn.setValue(true);
            metadataProvider.setMetadataTxnContext(mdTxnCtx.getValue());
        } else if (getDatasetType() == DatasetType.EXTERNAL) {
        }

        // #. finally, delete the dataset.
        MetadataManager.INSTANCE.dropDataset(mdTxnCtx.getValue(), dataverseName, datasetName, force);

        // drop inline types
        if (TypeUtil.isDatasetInlineTypeName(this, recordTypeDataverseName, recordTypeName)) {
            MetadataManager.INSTANCE.dropDatatype(mdTxnCtx.getValue(), recordTypeDataverseName, recordTypeName);
        }
        if (hasMetaPart() && TypeUtil.isDatasetInlineTypeName(this, metaTypeDataverseName, metaTypeName)) {
            MetadataManager.INSTANCE.dropDatatype(mdTxnCtx.getValue(), metaTypeDataverseName, metaTypeName);
        }

        // Drops the associated nodegroup if it is no longer used by any other dataset.
        if (dropCorrespondingNodeGroup) {
            metadataProvider.getApplicationContext().getMetadataLockManager()
                    .acquireNodeGroupWriteLock(metadataProvider.getLocks(), nodeGroupName);
            MetadataManager.INSTANCE.dropNodegroup(mdTxnCtx.getValue(), nodeGroupName, true);
        }
    }

    /**
     * Create the index dataflow helper factory for a particular index on the dataset
     *
     * @param mdProvider            metadata provider to get metadata information, components, and runtimes
     * @param index                 the index to get the dataflow helper factory for
     * @param recordType            the record type for the dataset
     * @param metaType              the meta type for the dataset
     * @param mergePolicyFactory    the merge policy factory of the dataset
     * @param mergePolicyProperties the merge policy properties for the dataset
     * @return indexDataflowHelperFactory
     * an instance of {@link org.apache.hyracks.storage.am.common.dataflow.IIndexDataflowHelperFactory}
     * @throws AlgebricksException if dataflow helper factory could not be created
     */
    public IResourceFactory getResourceFactory(MetadataProvider mdProvider, Index index, ARecordType recordType,
            ARecordType metaType, ILSMMergePolicyFactory mergePolicyFactory, Map<String, String> mergePolicyProperties)
            throws AlgebricksException {
        ITypeTraits[] filterTypeTraits = DatasetUtil.computeFilterTypeTraits(this, recordType, metaType);
        IBinaryComparatorFactory[] filterCmpFactories = DatasetUtil.computeFilterBinaryComparatorFactories(this,
                recordType, metaType, mdProvider.getStorageComponentProvider().getComparatorFactoryProvider());
        IResourceFactory resourceFactory;
        switch (index.getIndexType()) {
            case ARRAY:
                resourceFactory = ArrayBTreeResourceFactoryProvider.INSTANCE.getResourceFactory(mdProvider, this, index,
                        recordType, metaType, mergePolicyFactory, mergePolicyProperties, filterTypeTraits,
                        filterCmpFactories);
                break;
            case BTREE:
                resourceFactory = BTreeResourceFactoryProvider.INSTANCE.getResourceFactory(mdProvider, this, index,
                        recordType, metaType, mergePolicyFactory, mergePolicyProperties, filterTypeTraits,
                        filterCmpFactories);
                break;
            case RTREE:
                resourceFactory = rTreeResourceFactoryProvider.getResourceFactory(mdProvider, this, index, recordType,
                        metaType, mergePolicyFactory, mergePolicyProperties, filterTypeTraits, filterCmpFactories);
                break;
            case LENGTH_PARTITIONED_NGRAM_INVIX:
            case LENGTH_PARTITIONED_WORD_INVIX:
            case SINGLE_PARTITION_NGRAM_INVIX:
            case SINGLE_PARTITION_WORD_INVIX:
                resourceFactory = invertedIndexResourceFactoryProvider.getResourceFactory(mdProvider, this, index,
                        recordType, metaType, mergePolicyFactory, mergePolicyProperties, filterTypeTraits,
                        filterCmpFactories);
                break;
            case SAMPLE:
                resourceFactory = BTreeResourceFactoryProvider.INSTANCE.getResourceFactory(mdProvider, this, index,
                        recordType, metaType, mergePolicyFactory, mergePolicyProperties, null, null);
                break;
            default:
                throw new CompilationException(ErrorCode.COMPILATION_UNKNOWN_INDEX_TYPE,
                        index.getIndexType().toString());
        }
        return new DatasetLocalResourceFactory(datasetId, resourceFactory);
    }

    /**
     * Get the IO Operation callback factory for the index which belongs to this dataset
     *
     * @param index the index
     * @return ioOperationCallbackFactory
     * an instance of {@link org.apache.hyracks.storage.am.lsm.common.api.ILSMIOOperationCallbackFactory}
     * to be used with IO operations
     * @throws AlgebricksException if the factory could not be created for the index/dataset combination
     */
    @SuppressWarnings("squid:S1172")
    public ILSMIOOperationCallbackFactory getIoOperationCallbackFactory(Index index) throws AlgebricksException {
        if (isAtomic()) {
            return new AtomicLSMIndexIOOperationCallbackFactory(getComponentIdGeneratorFactory(),
                    getDatasetInfoProvider());
        } else {
            return new LSMIndexIOOperationCallbackFactory(getComponentIdGeneratorFactory(), getDatasetInfoProvider());
        }
    }

    public ILSMPageWriteCallbackFactory getPageWriteCallbackFactory() throws AlgebricksException {
        return new LSMIndexPageWriteCallbackFactory();
    }

    /**
     * get the IndexOperationTrackerFactory for a particular index on the dataset
     *
     * @param index the index
     * @return an instance of {@link org.apache.hyracks.storage.am.lsm.common.api.ILSMOperationTrackerFactory}
     */
    public ILSMOperationTrackerFactory getIndexOperationTrackerFactory(Index index) {
        return index.isPrimaryIndex() ? new PrimaryIndexOperationTrackerFactory(getDatasetId())
                : new SecondaryIndexOperationTrackerFactory(getDatasetId());
    }

    public ILSMComponentIdGeneratorFactory getComponentIdGeneratorFactory() {
        return new DatasetLSMComponentIdGeneratorFactory(getDatasetId());
    }

    public IDatasetInfoProvider getDatasetInfoProvider() {
        return new DatasetInfoProvider(getDatasetId());
    }

    public boolean isAtomic() {
        return datasetType == DatasetType.INTERNAL
                && ((InternalDatasetDetails) getDatasetDetails()).isDatasetWithoutTypeSpecification();
    }

    /**
     * Get search callback factory for this dataset with the passed index and operation
     *
     * @param storageComponentProvider         storage component provider
     * @param index                            the index
     * @param op                               the operation this search is part of
     * @param primaryKeyFields                 the primary key fields indexes for locking purposes
     * @param primaryKeyFieldsInSecondaryIndex the primary key fields indexes in the given secondary index (used for index-only plan)
     * @param proceedIndexOnlyPlan             the given plan is an index-only plan? (used for index-only plan)
     * @return an instance of {@link org.apache.hyracks.storage.am.common.api.ISearchOperationCallbackFactory}
     * @throws AlgebricksException if the callback factory could not be created
     */
    public ISearchOperationCallbackFactory getSearchCallbackFactory(IStorageComponentProvider storageComponentProvider,
            Index index, IndexOperation op, int[] primaryKeyFields, int[] primaryKeyFieldsInSecondaryIndex,
            boolean proceedIndexOnlyPlan) throws AlgebricksException {
        if (isAtomic()) {
            return NoOpOperationCallbackFactory.INSTANCE;
        }

        if (index.isPrimaryIndex()) {
            /**
             * Due to the read-committed isolation level,
             * we may acquire very short duration lock(i.e., instant lock) for readers.
             */
            return (op == IndexOperation.UPSERT || op == IndexOperation.INSERT)
                    ? new LockThenSearchOperationCallbackFactory(getDatasetId(), primaryKeyFields,
                            storageComponentProvider.getTransactionSubsystemProvider(), ResourceType.LSM_BTREE)
                    : new PrimaryIndexInstantSearchOperationCallbackFactory(getDatasetId(), primaryKeyFields,
                            storageComponentProvider.getTransactionSubsystemProvider(), ResourceType.LSM_BTREE);
        } else if (proceedIndexOnlyPlan) {
            // Index-only plan case: we need to execute instantTryLock on PK during a secondary-index search.
            // TODO: ResourceType is never used in the Callbackfactory. Should we keep it?
            return new SecondaryIndexInstanctSearchOperationCallbackFactory(getDatasetId(),
                    primaryKeyFieldsInSecondaryIndex, storageComponentProvider.getTransactionSubsystemProvider(),
                    index.resourceType());
        } else if (index.isPrimaryKeyIndex()) {
            // this is the case where the index is secondary primary index and locking is required
            // since the secondary primary index replaces the dataset index (which locks)
            return new PrimaryIndexInstantSearchOperationCallbackFactory(getDatasetId(), primaryKeyFields,
                    storageComponentProvider.getTransactionSubsystemProvider(), ResourceType.LSM_BTREE);
        }
        return new SecondaryIndexSearchOperationCallbackFactory();
    }

    /**
     * Get search callback factory for this dataset with the passed index and operation
     *
     * @param storageComponentProvider storage component provider
     * @param index                    the index
     * @param op                       the operation this search is part of
     * @param primaryKeyFields         the primary key fields indexes for locking purposes
     * @return an instance of {@link org.apache.hyracks.storage.am.common.api.ISearchOperationCallbackFactory}
     * @throws AlgebricksException if the callback factory could not be created
     */
    public ISearchOperationCallbackFactory getSearchCallbackFactory(IStorageComponentProvider storageComponentProvider,
            Index index, IndexOperation op, int[] primaryKeyFields) throws AlgebricksException {
        return getSearchCallbackFactory(storageComponentProvider, index, op, primaryKeyFields, null, false);
    }

    /**
     * Get the modification callback factory associated with this dataset, the passed index, and operation.
     *
     * @param index            the index
     * @param op               the operation performed for this callback
     * @param primaryKeyFields the indexes of the primary keys (used for lock operations)
     * @return an instance of {@link org.apache.hyracks.storage.am.common.api.IModificationOperationCallbackFactory}
     * @throws AlgebricksException If the callback factory could not be created
     */
    public IModificationOperationCallbackFactory getModificationCallbackFactory(
            IStorageComponentProvider componentProvider, Index index, IndexOperation op, int[] primaryKeyFields)
            throws AlgebricksException {
        if (isAtomic()) {
            return NoOpOperationCallbackFactory.INSTANCE;
        }

        if (index.isPrimaryIndex()) {
            return op == IndexOperation.UPSERT || op == IndexOperation.INSERT
                    ? new UpsertOperationCallbackFactory(getDatasetId(),
                            primaryKeyFields, componentProvider.getTransactionSubsystemProvider(), Operation.get(op),
                            index.resourceType())
                    : op == IndexOperation.DELETE ? new PrimaryIndexModificationOperationCallbackFactory(getDatasetId(),
                            primaryKeyFields, componentProvider.getTransactionSubsystemProvider(), Operation.get(op),
                            index.resourceType()) : NoOpOperationCallbackFactory.INSTANCE;
        } else {
            return op == IndexOperation.DELETE || op == IndexOperation.INSERT || op == IndexOperation.UPSERT
                    ? new SecondaryIndexModificationOperationCallbackFactory(getDatasetId(), primaryKeyFields,
                            componentProvider.getTransactionSubsystemProvider(), Operation.get(op),
                            index.resourceType())
                    : NoOpOperationCallbackFactory.INSTANCE;
        }
    }

    @Override
    public String toString() {
        ObjectMapper mapper = new ObjectMapper();
        try {
            return mapper.writeValueAsString(toMap());
        } catch (JsonProcessingException e) {
            LOGGER.log(Level.WARN, "Unable to convert map to json String", e);
            return dataverseName + "." + datasetName;
        }
    }

    public Map<String, Object> toMap() {
        Map<String, Object> tree = new HashMap<>();
        tree.put("datasetId", Integer.toString(datasetId));
        tree.put("dataverseName", dataverseName);
        tree.put("datasetName", datasetName);
        tree.put("recordTypeDataverseName", recordTypeDataverseName);
        tree.put("recordTypeName", recordTypeName);
        tree.put("nodeGroupName", nodeGroupName);
        tree.put("compactionPolicyFactory", compactionPolicyFactory);
        tree.put("hints", hints);
        tree.put("compactionPolicyProperties", compactionPolicyProperties);
        tree.put("datasetType", datasetType.name());
        tree.put("datasetDetails", datasetDetails.toString());
        tree.put("metaTypeDataverseName", metaTypeDataverseName);
        tree.put("metaTypeName", metaTypeName);
        tree.put("pendingOp", MetadataUtil.pendingOpToString(pendingOp));
        tree.put("rebalanceCount", rebalanceCount);
        tree.put("compressionScheme", compressionScheme);
        tree.put("datasetFormatInfo", datasetFormatInfo.toString());
        return tree;
    }

    @Override
    public int hashCode() {
        return Objects.hash(dataverseName, datasetName);
    }

    /**
     * Gets the commit runtime factory for inserting/upserting/deleting operations on this dataset.
     *
     * @param metadataProvider,           the metadata provider.
     * @param primaryKeyFieldPermutation, the primary key field permutation according to the input.
     * @param isSink,                     whether this commit runtime is the last operator in the pipeline.
     * @return the commit runtime factory for inserting/upserting/deleting operations on this dataset.
     * @throws AlgebricksException
     */
    public IPushRuntimeFactory getCommitRuntimeFactory(MetadataProvider metadataProvider,
            int[] primaryKeyFieldPermutation, boolean isSink) throws AlgebricksException {
        int[] datasetPartitions = getDatasetPartitions(metadataProvider);
        IBinaryHashFunctionFactory[] pkHashFunFactories = getPrimaryHashFunctionFactories(metadataProvider);
        ITuplePartitionerFactory partitionerFactory = new FieldHashPartitionerFactory(primaryKeyFieldPermutation,
                pkHashFunFactories, datasetPartitions.length);
        if (isAtomic()) {
            return new NoOpCommitRuntimeFactory(datasetId, primaryKeyFieldPermutation,
                    metadataProvider.isWriteTransaction(), datasetPartitions, isSink, partitionerFactory);
        }
        return new CommitRuntimeFactory(datasetId, primaryKeyFieldPermutation, metadataProvider.isWriteTransaction(),
                datasetPartitions, isSink, partitionerFactory);
    }

    public IFrameOperationCallbackFactory getFrameOpCallbackFactory(MetadataProvider mdProvider) {
        return NoOpFrameOperationCallbackFactory.INSTANCE;
    }

    public boolean isCorrelated() {
        return CorrelatedPrefixMergePolicyFactory.NAME.equals(compactionPolicyFactory);
    }

    @Override
    public List<List<String>> getPrimaryKeys() {
        if (getDatasetType() == DatasetType.EXTERNAL) {
            return Collections.emptyList();
        }
        return ((InternalDatasetDetails) getDatasetDetails()).getPartitioningKey();
    }

    /**
     * @return the array of type traits that contains the following type traits in order
     * 1) the primary keys,
     * 2) the query record type,
     * 3) the metadata type trait if the dataset has metadata
     */
    // ToDo: returning such an array can be confusing because it may contain the metadata type or not.
    // instead of returning an array, create a new class that contains 1) a type trait array for the primary keys,
    // 2) the record type trait, and 3) an nullable meta type trait
    public ITypeTraits[] getPrimaryTypeTraits(MetadataProvider metadataProvider, ARecordType recordType,
            ARecordType metaType) throws AlgebricksException {
        IStorageComponentProvider storageComponentProvider = metadataProvider.getStorageComponentProvider();
        ITypeTraitProvider ttProvider = storageComponentProvider.getTypeTraitProvider();
        List<List<String>> partitioningKeys = getPrimaryKeys();
        int numPrimaryKeys = partitioningKeys.size();
        ITypeTraits[] typeTraits = new ITypeTraits[numPrimaryKeys + 1 + (hasMetaPart() ? 1 : 0)];
        List<Integer> indicators = null;
        if (hasMetaPart()) {
            indicators = ((InternalDatasetDetails) getDatasetDetails()).getKeySourceIndicator();
        }
        for (int i = 0; i < numPrimaryKeys; i++) {
            IAType keyType = datasetType == DatasetType.EXTERNAL ? null
                    : (indicators == null || indicators.get(i) == 0)
                            ? recordType.getSubFieldType(partitioningKeys.get(i))
                            : metaType.getSubFieldType(partitioningKeys.get(i));
            typeTraits[i] = ttProvider.getTypeTrait(keyType);
        }
        typeTraits[numPrimaryKeys] = ttProvider.getTypeTrait(recordType);
        if (hasMetaPart()) {
            typeTraits[numPrimaryKeys + 1] = ttProvider.getTypeTrait(metaType);
        }
        return typeTraits;
    }

    /**
     * Gets the record descriptor for primary records of this dataset.
     *
     * @param metadataProvider, the metadata provider.
     * @return the record descriptor for primary records of this dataset.
     * @throws AlgebricksException
     */
    public RecordDescriptor getPrimaryRecordDescriptor(MetadataProvider metadataProvider) throws AlgebricksException {
        List<List<String>> partitioningKeys = getPrimaryKeys();
        int numPrimaryKeys = partitioningKeys.size();
        ISerializerDeserializer[] primaryRecFields =
                new ISerializerDeserializer[numPrimaryKeys + 1 + (hasMetaPart() ? 1 : 0)];
        ITypeTraits[] primaryTypeTraits = new ITypeTraits[numPrimaryKeys + 1 + (hasMetaPart() ? 1 : 0)];
        ISerializerDeserializerProvider serdeProvider = metadataProvider.getDataFormat().getSerdeProvider();
        List<Integer> indicators = null;
        if (hasMetaPart()) {
            indicators = ((InternalDatasetDetails) getDatasetDetails()).getKeySourceIndicator();
        }
        ARecordType itemType = (ARecordType) metadataProvider.findType(this);
        ARecordType metaType = (ARecordType) metadataProvider.findMetaType(this);
        itemType = (ARecordType) metadataProvider.findTypeForDatasetWithoutType(itemType, metaType, this);

        // Set the serde/traits for primary keys
        for (int i = 0; i < numPrimaryKeys; i++) {
            IAType keyType =
                    datasetType == DatasetType.EXTERNAL ? null
                            : (indicators == null || indicators.get(i) == 0)
                                    ? itemType.getSubFieldType(partitioningKeys.get(i))
                                    : metaType.getSubFieldType(partitioningKeys.get(i));
            primaryRecFields[i] = serdeProvider.getSerializerDeserializer(keyType);
            primaryTypeTraits[i] = TypeTraitProvider.INSTANCE.getTypeTrait(keyType);
        }

        // Set the serde for the record field
        primaryRecFields[numPrimaryKeys] = SerializerDeserializerProvider.INSTANCE.getSerializerDeserializer(itemType);
        primaryTypeTraits[numPrimaryKeys] = TypeTraitProvider.INSTANCE.getTypeTrait(itemType);
        if (hasMetaPart()) {
            // Set the serde and traits for the meta record field
            primaryRecFields[numPrimaryKeys + 1] =
                    SerializerDeserializerProvider.INSTANCE.getSerializerDeserializer(metaType);
            primaryTypeTraits[numPrimaryKeys + 1] = TypeTraitProvider.INSTANCE.getTypeTrait(itemType);
        }
        return new RecordDescriptor(primaryRecFields, primaryTypeTraits);
    }

    /**
     * Gets the comparator factories for the primary key fields of this dataset.
     *
     * @param metadataProvider, the metadata provider.
     * @return the comparator factories for the primary key fields of this dataset.
     * @throws AlgebricksException
     */
    public IBinaryComparatorFactory[] getPrimaryComparatorFactories(MetadataProvider metadataProvider,
            ARecordType recordType, ARecordType metaType) throws AlgebricksException {
        IStorageComponentProvider storageComponentProvider = metadataProvider.getStorageComponentProvider();
        IBinaryComparatorFactoryProvider cmpFactoryProvider = storageComponentProvider.getComparatorFactoryProvider();
        List<List<String>> partitioningKeys = getPrimaryKeys();
        int numPrimaryKeys = partitioningKeys.size();
        IBinaryComparatorFactory[] cmpFactories = new IBinaryComparatorFactory[numPrimaryKeys];
        List<Integer> indicators = null;
        if (hasMetaPart()) {
            indicators = ((InternalDatasetDetails) getDatasetDetails()).getKeySourceIndicator();
        }
        for (int i = 0; i < numPrimaryKeys; i++) {
            IAType keyType = datasetType == DatasetType.EXTERNAL ? null
                    : (indicators == null || indicators.get(i) == 0)
                            ? recordType.getSubFieldType(partitioningKeys.get(i))
                            : metaType.getSubFieldType(partitioningKeys.get(i));
            cmpFactories[i] = cmpFactoryProvider.getBinaryComparatorFactory(keyType, true);
        }
        return cmpFactories;
    }

    /**
     * Gets the hash function factories for the primary key fields of this dataset.
     *
     * @param metadataProvider, the metadata provider.
     * @return the hash function factories for the primary key fields of this dataset.
     * @throws AlgebricksException
     */
    public IBinaryHashFunctionFactory[] getPrimaryHashFunctionFactories(MetadataProvider metadataProvider)
            throws AlgebricksException {
        ARecordType recordType = (ARecordType) metadataProvider.findType(this);
        ARecordType metaType = (ARecordType) metadataProvider.findMetaType(this);
        recordType = (ARecordType) metadataProvider.findTypeForDatasetWithoutType(recordType, metaType, this);

        List<List<String>> partitioningKeys = getPrimaryKeys();
        int numPrimaryKeys = partitioningKeys.size();
        IBinaryHashFunctionFactory[] hashFuncFactories = new IBinaryHashFunctionFactory[numPrimaryKeys];
        List<Integer> indicators = null;
        if (hasMetaPart()) {
            indicators = ((InternalDatasetDetails) getDatasetDetails()).getKeySourceIndicator();
        }
        for (int i = 0; i < numPrimaryKeys; i++) {
            IAType keyType = datasetType == DatasetType.EXTERNAL ? null
                    : (indicators == null || indicators.get(i) == 0)
                            ? recordType.getSubFieldType(partitioningKeys.get(i))
                            : metaType.getSubFieldType(partitioningKeys.get(i));
            hashFuncFactories[i] = BinaryHashFunctionFactoryProvider.INSTANCE.getBinaryHashFunctionFactory(keyType);
        }
        return hashFuncFactories;
    }

    @Override
    public int[] getPrimaryBloomFilterFields() {
        List<List<String>> partitioningKeys = getPrimaryKeys();
        int numPrimaryKeys = partitioningKeys.size();
        return IntStream.range(0, numPrimaryKeys).toArray();
    }

    // Gets the target dataset for the purpose of rebalance.
    public Dataset getTargetDatasetForRebalance(String targetNodeGroupName) {
        return new Dataset(this.dataverseName, this.datasetName, this.recordTypeDataverseName, this.recordTypeName,
                this.metaTypeDataverseName, this.metaTypeName, targetNodeGroupName, this.compactionPolicyFactory,
                this.compactionPolicyProperties, this.datasetDetails, this.hints, this.datasetType,
                DatasetIdFactory.generateAlternatingDatasetId(this.datasetId), this.pendingOp, this.rebalanceCount + 1,
                this.compressionScheme, this.datasetFormatInfo);
    }

    // Gets an array of partition numbers for this dataset.
    protected int[] getDatasetPartitions(MetadataProvider metadataProvider) throws AlgebricksException {
        FileSplit[] splitsForDataset =
                metadataProvider.splitsForIndex(metadataProvider.getMetadataTxnContext(), this, getDatasetName());
        int[] partitions = new int[splitsForDataset.length];
        for (int i = 0; i < partitions.length; i++) {
            partitions[i] = StoragePathUtil.getPartitionNumFromRelativePath(splitsForDataset[i].getPath());
        }
        return partitions;
    }

    public String getCompressionScheme() {
        return compressionScheme;
    }

    public DatasetFullyQualifiedName getDatasetFullyQualifiedName() {
        return datasetFullyQualifiedName;
    }

    public DatasetFormatInfo getDatasetFormatInfo() {
        return datasetFormatInfo;
    }
}
