/*
 * 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.cassandra.index.internal;

import java.util.List;

import org.apache.cassandra.config.CFMetaData;
import org.apache.cassandra.config.ColumnDefinition;
import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.marshal.AbstractType;
import org.apache.cassandra.db.marshal.CollectionType;
import org.apache.cassandra.db.marshal.CompositeType;
import org.apache.cassandra.index.internal.composites.*;
import org.apache.cassandra.index.internal.keys.KeysIndex;
import org.apache.cassandra.schema.IndexMetadata;

public interface CassandraIndexFunctions
{
    /**
     *
     * @param baseCfs
     * @param indexMetadata
     * @return
     */
    public CassandraIndex newIndexInstance(ColumnFamilyStore baseCfs, IndexMetadata indexMetadata);

    /**
     * Returns the type of the the values in the index. For most columns this is simply its type, but for collections
     * it depends on whether the index is on the collection name/value element or on a frozen collection
     * @param indexedColumn
     * @return
     */
    default AbstractType<?> getIndexedValueType(ColumnDefinition indexedColumn)
    {
        return indexedColumn.type;
    }

    /**
     * Add the clustering columns for a specific type of index table to the a CFMetaData.Builder (which is being
     * used to construct the index table's CFMetadata. In the default implementation, the clustering columns of the
     * index table hold the partition key and clustering columns of the base table. This is overridden in several cases:
     * * When the indexed value is itself a clustering column, in which case, we only need store the base table's
     *   *other* clustering values in the index - the indexed value being the index table's partition key
     * * When the indexed value is a collection value, in which case we also need to capture the cell path from the base
     *   table
     * * In a KEYS index (for thrift/compact storage/static column indexes), where only the base partition key is
     *   held in the index table.
     *
     * Called from indexCfsMetadata
     * @param builder
     * @param baseMetadata
     * @param cfDef
     * @return
     */
    default CFMetaData.Builder addIndexClusteringColumns(CFMetaData.Builder builder,
                                                         CFMetaData baseMetadata,
                                                         ColumnDefinition cfDef)
    {
        for (ColumnDefinition def : baseMetadata.clusteringColumns())
            builder.addClusteringColumn(def.name, def.type);
        return builder;
    }

    /*
     * implementations providing specializations for the built in index types
     */

    static final CassandraIndexFunctions KEYS_INDEX_FUNCTIONS = new CassandraIndexFunctions()
    {
        public CassandraIndex newIndexInstance(ColumnFamilyStore baseCfs, IndexMetadata indexMetadata)
        {
            return new KeysIndex(baseCfs, indexMetadata);
        }
    };

    static final CassandraIndexFunctions REGULAR_COLUMN_INDEX_FUNCTIONS = new CassandraIndexFunctions()
    {
        public CassandraIndex newIndexInstance(ColumnFamilyStore baseCfs, IndexMetadata indexMetadata)
        {
            return new RegularColumnIndex(baseCfs, indexMetadata);
        }
    };

    static final CassandraIndexFunctions CLUSTERING_COLUMN_INDEX_FUNCTIONS = new CassandraIndexFunctions()
    {
        public CassandraIndex newIndexInstance(ColumnFamilyStore baseCfs, IndexMetadata indexMetadata)
        {
            return new ClusteringColumnIndex(baseCfs, indexMetadata);
        }

        public CFMetaData.Builder addIndexClusteringColumns(CFMetaData.Builder builder,
                                                            CFMetaData baseMetadata,
                                                            ColumnDefinition columnDef)
        {
            List<ColumnDefinition> cks = baseMetadata.clusteringColumns();
            for (int i = 0; i < columnDef.position(); i++)
            {
                ColumnDefinition def = cks.get(i);
                builder.addClusteringColumn(def.name, def.type);
            }
            for (int i = columnDef.position() + 1; i < cks.size(); i++)
            {
                ColumnDefinition def = cks.get(i);
                builder.addClusteringColumn(def.name, def.type);
            }
            return builder;
        }
    };

    static final CassandraIndexFunctions PARTITION_KEY_INDEX_FUNCTIONS = new CassandraIndexFunctions()
    {
        public CassandraIndex newIndexInstance(ColumnFamilyStore baseCfs, IndexMetadata indexMetadata)
        {
            return new PartitionKeyIndex(baseCfs, indexMetadata);
        }
    };

    static final CassandraIndexFunctions COLLECTION_KEY_INDEX_FUNCTIONS = new CassandraIndexFunctions()
    {
        public CassandraIndex newIndexInstance(ColumnFamilyStore baseCfs, IndexMetadata indexMetadata)
        {
            return new CollectionKeyIndex(baseCfs, indexMetadata);
        }

        public AbstractType<?> getIndexedValueType(ColumnDefinition indexedColumn)
        {
            return ((CollectionType) indexedColumn.type).nameComparator();
        }
    };

    static final CassandraIndexFunctions COLLECTION_VALUE_INDEX_FUNCTIONS = new CassandraIndexFunctions()
    {

        public CassandraIndex newIndexInstance(ColumnFamilyStore baseCfs, IndexMetadata indexMetadata)
        {
            return new CollectionValueIndex(baseCfs, indexMetadata);
        }

        public AbstractType<?> getIndexedValueType(ColumnDefinition indexedColumn)
        {
            return ((CollectionType)indexedColumn.type).valueComparator();
        }

        public CFMetaData.Builder addIndexClusteringColumns(CFMetaData.Builder builder,
                                                            CFMetaData baseMetadata,
                                                            ColumnDefinition columnDef)
        {
            for (ColumnDefinition def : baseMetadata.clusteringColumns())
                builder.addClusteringColumn(def.name, def.type);

            // collection key
            builder.addClusteringColumn("cell_path", ((CollectionType)columnDef.type).nameComparator());
            return builder;
        }
    };

    static final CassandraIndexFunctions COLLECTION_ENTRY_INDEX_FUNCTIONS = new CassandraIndexFunctions()
    {
        public CassandraIndex newIndexInstance(ColumnFamilyStore baseCfs, IndexMetadata indexMetadata)
        {
            return new CollectionEntryIndex(baseCfs, indexMetadata);
        }

        public AbstractType<?> getIndexedValueType(ColumnDefinition indexedColumn)
        {
            CollectionType colType = (CollectionType)indexedColumn.type;
            return CompositeType.getInstance(colType.nameComparator(), colType.valueComparator());
        }
    };
}