/*
 * 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.io.sstable;

import java.io.File;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.charset.CharacterCodingException;
import java.nio.file.Files;
import java.nio.file.Path;
import java.util.ArrayList;
import java.util.EnumSet;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.function.Supplier;
import java.util.stream.Collectors;
import java.util.stream.Stream;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import org.apache.cassandra.config.CFMetaData;
import org.apache.cassandra.config.ColumnDefinition;
import org.apache.cassandra.config.Schema;
import org.apache.cassandra.config.SchemaConstants;
import org.apache.cassandra.cql3.CQL3Type;
import org.apache.cassandra.cql3.statements.IndexTarget;
import org.apache.cassandra.db.Directories;
import org.apache.cassandra.db.SerializationHeader;
import org.apache.cassandra.db.lifecycle.LifecycleTransaction;
import org.apache.cassandra.db.marshal.AbstractCompositeType;
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.db.marshal.DynamicCompositeType;
import org.apache.cassandra.db.marshal.ListType;
import org.apache.cassandra.db.marshal.MapType;
import org.apache.cassandra.db.marshal.SetType;
import org.apache.cassandra.db.marshal.TupleType;
import org.apache.cassandra.db.marshal.UserType;
import org.apache.cassandra.io.sstable.metadata.MetadataComponent;
import org.apache.cassandra.io.sstable.metadata.MetadataType;
import org.apache.cassandra.schema.IndexMetadata;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.CassandraVersion;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.Pair;

/**
 * Validates and fixes type issues in the serialization-header of sstables.
 */
public abstract class SSTableHeaderFix
{
    // C* 3.0 upgrade code

    private static final String SKIPAUTOMATICUDTFIX = "cassandra.skipautomaticudtfix";
    private static final boolean SKIP_AUTOMATIC_FIX_ON_UPGRADE = Boolean.getBoolean(SKIPAUTOMATICUDTFIX);

    public static void fixNonFrozenUDTIfUpgradeFrom30()
    {
        String previousVersionString = FBUtilities.getPreviousReleaseVersionString();
        if (previousVersionString == null)
            return;
        CassandraVersion previousVersion = new CassandraVersion(previousVersionString);
        if (previousVersion.major != 3 || previousVersion.minor > 0)
        {
            // Not an upgrade from 3.0 to 3.x, nothing to do here
            return;
        }

        if (SKIP_AUTOMATIC_FIX_ON_UPGRADE)
        {
            logger.warn("Detected upgrade from {} to {}, but -D{}=true, NOT fixing UDT type references in " +
                        "sstable metadata serialization-headers",
                        previousVersionString,
                        FBUtilities.getReleaseVersionString(),
                        SKIPAUTOMATICUDTFIX);
            return;
        }

        logger.info("Detected upgrade from {} to {}, fixing UDT type references in sstable metadata serialization-headers",
                    previousVersionString,
                    FBUtilities.getReleaseVersionString());

        SSTableHeaderFix instance = SSTableHeaderFix.builder()
                                                    .schemaCallback(() -> Schema.instance::getCFMetaData)
                                                    .build();
        instance.execute();
    }

    // "regular" SSTableHeaderFix code, also used by StandaloneScrubber.

    private static final Logger logger = LoggerFactory.getLogger(SSTableHeaderFix.class);

    protected final Consumer<String> info;
    protected final Consumer<String> warn;
    protected final Consumer<String> error;
    protected final boolean dryRun;
    protected final Function<Descriptor, CFMetaData> schemaCallback;

    private final List<Descriptor> descriptors;

    private final List<Pair<Descriptor, Map<MetadataType, MetadataComponent>>> updates = new ArrayList<>();
    private boolean hasErrors;

    SSTableHeaderFix(Builder builder)
    {
        this.info = builder.info;
        this.warn = builder.warn;
        this.error = builder.error;
        this.dryRun = builder.dryRun;
        this.schemaCallback = builder.schemaCallback.get();
        this.descriptors = new ArrayList<>(builder.descriptors);
        Objects.requireNonNull(this.info, "info is null");
        Objects.requireNonNull(this.warn, "warn is null");
        Objects.requireNonNull(this.error, "error is null");
        Objects.requireNonNull(this.schemaCallback, "schemaCallback is null");
    }

    public static Builder builder()
    {
        return new Builder();
    }

    /**
     * Builder to configure and construct an instance of {@link SSTableHeaderFix}.
     * Default settings:
     * <ul>
     *     <li>log via the slf4j logger of {@link SSTableHeaderFix}</li>
     *     <li>no dry-run (i.e. validate and fix, if no serious errors are detected)</li>
     *     <li>no schema callback</li>
     * </ul>
     * If neither {@link #withDescriptor(Descriptor)} nor {@link #withPath(Path)} are used,
     * all "live" sstables in all data directories will be scanned.
     */
    public static class Builder
    {
        private final List<Path> paths = new ArrayList<>();
        private final List<Descriptor> descriptors = new ArrayList<>();
        private Consumer<String> info = (ln) -> logger.info("{}", ln);
        private Consumer<String> warn = (ln) -> logger.warn("{}", ln);
        private Consumer<String> error = (ln) -> logger.error("{}", ln);
        private boolean dryRun;
        private Supplier<Function<Descriptor, CFMetaData>> schemaCallback = () -> null;

        private Builder()
        {}

        /**
         * Only validate and prepare fix, but do not write updated (fixed) sstable serialization-headers.
         */
        public Builder dryRun()
        {
            dryRun = true;
            return this;
        }

        public Builder info(Consumer<String> output)
        {
            this.info = output;
            return this;
        }

        public Builder warn(Consumer<String> warn)
        {
            this.warn = warn;
            return this;
        }

        public Builder error(Consumer<String> error)
        {
            this.error = error;
            return this;
        }

        /**
         * Manually provide an individual sstable or directory containing sstables.
         *
         * Implementation note: procesing "live" sstables in their data directories as well as sstables
         * in snapshots and backups in the data directories works.
         *
         * But processing sstables that reside somewhere else (i.e. verifying sstables before import)
         * requires the use of {@link #withDescriptor(Descriptor)}.
         */
        public Builder withPath(Path path)
        {
            this.paths.add(path);
            return this;
        }

        public Builder withDescriptor(Descriptor descriptor)
        {
            this.descriptors.add(descriptor);
            return this;
        }

        /**
         * Schema callback to retrieve the schema of a table. Production code always delegates to the
         * live schema ({@code Schema.instance}). Unit tests use this method to feed a custom schema.
         */
        public Builder schemaCallback(Supplier<Function<Descriptor, CFMetaData>> schemaCallback)
        {
            this.schemaCallback = schemaCallback;
            return this;
        }

        public SSTableHeaderFix build()
        {
            if (paths.isEmpty() && descriptors.isEmpty())
                return new AutomaticHeaderFix(this);

            return new ManualHeaderFix(this);
        }

        public Builder logToList(List<String> output)
        {
            return info(ln -> output.add("INFO  " + ln))
                   .warn(ln -> output.add("WARN  " + ln))
                   .error(ln -> output.add("ERROR " + ln));
        }
    }

    public final void execute()
    {
        prepare();

        logger.debug("Processing {} sstables:{}",
                     descriptors.size(),
                     descriptors.stream().map(Descriptor::toString).collect(Collectors.joining("\n    ", "\n    ", "")));

        descriptors.forEach(this::processSSTable);

        if (updates.isEmpty())
            return;

        if (hasErrors)
        {
            info.accept("Stopping due to previous errors. Either fix the errors or specify the ignore-errors option.");
            return;
        }

        if (dryRun)
        {
            info.accept("Not fixing identified and fixable serialization-header issues.");
            return;
        }

        info.accept("Writing new metadata files");
        updates.forEach(descAndMeta -> writeNewMetadata(descAndMeta.left, descAndMeta.right));
        info.accept("Finished writing new metadata files");
    }

    /**
     * Whether {@link #execute()} encountered an error.
     */
    public boolean hasError()
    {
        return hasErrors;
    }

    /**
     * Whether {@link #execute()} found mismatches.
     */
    public boolean hasChanges()
    {
        return !updates.isEmpty();
    }

    abstract void prepare();

    private void error(String format, Object... args)
    {
        hasErrors = true;
        error.accept(String.format(format, args));
    }

    void processFileOrDirectory(Path path)
    {
        Stream.of(path)
              .flatMap(SSTableHeaderFix::maybeExpandDirectory)
              .filter(p -> {
                  File f = p.toFile();
                  return Component.fromFilename(f.getParentFile(), f.getName()).right.type == Component.Type.DATA;
              })
              .map(Path::toString)
              .map(Descriptor::fromFilename)
              .forEach(descriptors::add);
    }

    private static Stream<Path> maybeExpandDirectory(Path path)
    {
        if (Files.isRegularFile(path))
            return Stream.of(path);
        return LifecycleTransaction.getFiles(path, (file, fileType) -> fileType == Directories.FileType.FINAL, Directories.OnTxnErr.IGNORE)
                                   .stream()
                                   .map(File::toPath);
    }

    private void processSSTable(Descriptor desc)
    {
        if (desc.cfname.indexOf('.') != -1)
        {
            // secondary index not checked

            // partition-key is the indexed column type
            // clustering-key is org.apache.cassandra.db.marshal.PartitionerDefinedOrder
            // no static columns, no regular columns
            return;
        }

        CFMetaData tableMetadata = schemaCallback.apply(desc);
        if (tableMetadata == null)
        {
            error("Table %s.%s not found in the schema - NOT checking sstable %s", desc.ksname, desc.cfname, desc);
            return;
        }

        Set<Component> components = SSTable.discoverComponentsFor(desc);
        if (components.stream().noneMatch(c -> c.type == Component.Type.STATS))
        {
            error("sstable %s has no -Statistics.db component.", desc);
            return;
        }

        Map<MetadataType, MetadataComponent> metadata = readSSTableMetadata(desc);
        if (metadata == null)
            return;

        MetadataComponent component = metadata.get(MetadataType.HEADER);
        if (!(component instanceof SerializationHeader.Component))
        {
            error("sstable %s: Expected %s, but got %s from metadata.get(MetadataType.HEADER)",
                  desc,
                  SerializationHeader.Component.class.getName(),
                  component != null ? component.getClass().getName() : "'null'");
            return;
        }
        SerializationHeader.Component header = (SerializationHeader.Component) component;

        // check partition key type
        AbstractType<?> keyType = validatePartitionKey(desc, tableMetadata, header);

        // check clustering columns
        List<AbstractType<?>> clusteringTypes = validateClusteringColumns(desc, tableMetadata, header);

        // check static and regular columns
        Map<ByteBuffer, AbstractType<?>> staticColumns = validateColumns(desc, tableMetadata, header.getStaticColumns(), ColumnDefinition.Kind.STATIC);
        Map<ByteBuffer, AbstractType<?>> regularColumns = validateColumns(desc, tableMetadata, header.getRegularColumns(), ColumnDefinition.Kind.REGULAR);

        SerializationHeader.Component newHeader = SerializationHeader.Component.buildComponentForTools(keyType,
                                                                                                       clusteringTypes,
                                                                                                       staticColumns,
                                                                                                       regularColumns,
                                                                                                       header.getEncodingStats());

        // SerializationHeader.Component has no equals(), but a "good" toString()
        if (header.toString().equals(newHeader.toString()))
            return;

        Map<MetadataType, MetadataComponent> newMetadata = new LinkedHashMap<>(metadata);
        newMetadata.put(MetadataType.HEADER, newHeader);

        updates.add(Pair.create(desc, newMetadata));
    }

    private AbstractType<?> validatePartitionKey(Descriptor desc, CFMetaData tableMetadata, SerializationHeader.Component header)
    {
        boolean keyMismatch = false;
        AbstractType<?> headerKeyType = header.getKeyType();
        AbstractType<?> schemaKeyType = tableMetadata.getKeyValidator();
        boolean headerKeyComposite = headerKeyType instanceof CompositeType;
        boolean schemaKeyComposite = schemaKeyType instanceof CompositeType;
        if (headerKeyComposite != schemaKeyComposite)
        {
            // one is a composite partition key, the other is not - very suspicious
            keyMismatch = true;
        }
        else if (headerKeyComposite) // && schemaKeyComposite
        {
            // Note, the logic is similar as just calling 'fixType()' using the composite partition key,
            // but the log messages should use the composite partition key column names.
            List<AbstractType<?>> headerKeyComponents = ((CompositeType) headerKeyType).types;
            List<AbstractType<?>> schemaKeyComponents = ((CompositeType) schemaKeyType).types;
            if (headerKeyComponents.size() != schemaKeyComponents.size())
            {
                // different number of components in composite partition keys - very suspicious
                keyMismatch = true;
                // Just use the original type from the header. Since the number of partition key components
                // don't match, there's nothing to meaningfully validate against.
            }
            else
            {
                // fix components in composite partition key, if necessary
                List<AbstractType<?>> newComponents = new ArrayList<>(schemaKeyComponents.size());
                for (int i = 0; i < schemaKeyComponents.size(); i++)
                {
                    AbstractType<?> headerKeyComponent = headerKeyComponents.get(i);
                    AbstractType<?> schemaKeyComponent = schemaKeyComponents.get(i);
                    AbstractType<?> fixedType = fixType(desc,
                                                        tableMetadata.partitionKeyColumns().get(i).name.bytes,
                                                        headerKeyComponent,
                                                        schemaKeyComponent,
                                                        false);
                    if (fixedType == null)
                        keyMismatch = true;
                    else
                        headerKeyComponent = fixedType;
                    newComponents.add(fixType(desc,
                                              tableMetadata.partitionKeyColumns().get(i).name.bytes,
                                              headerKeyComponent,
                                              schemaKeyComponent,
                                              false));
                }
                headerKeyType = CompositeType.getInstance(newComponents);
            }
        }
        else
        {
            // fix non-composite partition key, if necessary
            AbstractType<?> fixedType = fixType(desc, tableMetadata.partitionKeyColumns().get(0).name.bytes, headerKeyType, schemaKeyType, false);
            if (fixedType == null)
                // non-composite partition key doesn't match and cannot be fixed
                keyMismatch = true;
            else
                headerKeyType = fixedType;
        }
        if (keyMismatch)
            error("sstable %s: Mismatch in partition key type between sstable serialization-header and schema (%s vs %s)",
                  desc,
                  headerKeyType.asCQL3Type(),
                  schemaKeyType.asCQL3Type());
        return headerKeyType;
    }

    private List<AbstractType<?>> validateClusteringColumns(Descriptor desc, CFMetaData tableMetadata, SerializationHeader.Component header)
    {
        List<AbstractType<?>> headerClusteringTypes = header.getClusteringTypes();
        List<AbstractType<?>> clusteringTypes = new ArrayList<>();
        boolean clusteringMismatch = false;
        List<ColumnDefinition> schemaClustering = tableMetadata.clusteringColumns();
        if (schemaClustering.size() != headerClusteringTypes.size())
        {
            clusteringMismatch = true;
            // Just use the original types. Since the number of clustering columns don't match, there's nothing to
            // meaningfully validate against.
            clusteringTypes.addAll(headerClusteringTypes);
        }
        else
        {
            for (int i = 0; i < headerClusteringTypes.size(); i++)
            {
                AbstractType<?> headerType = headerClusteringTypes.get(i);
                ColumnDefinition column = schemaClustering.get(i);
                AbstractType<?> schemaType = column.type;
                AbstractType<?> fixedType = fixType(desc, column.name.bytes, headerType, schemaType, false);
                if (fixedType == null)
                    clusteringMismatch = true;
                else
                    headerType = fixedType;
                clusteringTypes.add(headerType);
            }
        }
        if (clusteringMismatch)
            error("sstable %s: mismatch in clustering columns between sstable serialization-header and schema (%s vs %s)",
                  desc,
                  headerClusteringTypes.stream().map(AbstractType::asCQL3Type).map(CQL3Type::toString).collect(Collectors.joining(",")),
                  schemaClustering.stream().map(cd -> cd.type.asCQL3Type().toString()).collect(Collectors.joining(",")));
        return clusteringTypes;
    }

    private Map<ByteBuffer, AbstractType<?>> validateColumns(Descriptor desc, CFMetaData tableMetadata, Map<ByteBuffer, AbstractType<?>> columns, ColumnDefinition.Kind kind)
    {
        Map<ByteBuffer, AbstractType<?>> target = new LinkedHashMap<>();
        for (Map.Entry<ByteBuffer, AbstractType<?>> nameAndType : columns.entrySet())
        {
            ByteBuffer name = nameAndType.getKey();
            AbstractType<?> type = nameAndType.getValue();

            AbstractType<?> fixedType = validateColumn(desc, tableMetadata, kind, name, type);
            if (fixedType == null)
            {
                error("sstable %s: contains column '%s' of type '%s', which could not be validated",
                      desc,
                      type,
                      logColumnName(name));
                // don't use a "null" type instance
                fixedType = type;
            }

            target.put(name, fixedType);
        }
        return target;
    }

    private AbstractType<?> validateColumn(Descriptor desc, CFMetaData tableMetadata, ColumnDefinition.Kind kind, ByteBuffer name, AbstractType<?> type)
    {
        ColumnDefinition cd = tableMetadata.getColumnDefinition(name);
        if (cd == null)
        {
            // In case the column was dropped, there is not much that we can actually validate.
            // The column could have been recreated using the same or a different kind or the same or
            // a different type. Lottery...

            cd = tableMetadata.getDroppedColumnDefinition(name, kind == ColumnDefinition.Kind.STATIC);
            if (cd == null)
            {
                for (IndexMetadata indexMetadata : tableMetadata.getIndexes())
                {
                    String target = indexMetadata.options.get(IndexTarget.TARGET_OPTION_NAME);
                    if (target != null && ByteBufferUtil.bytes(target).equals(name))
                    {
                        warn.accept(String.format("sstable %s: contains column '%s', which is not a column in the table '%s.%s', but a target for that table's index '%s'",
                                                  desc,
                                                  logColumnName(name),
                                                  tableMetadata.ksName,
                                                  tableMetadata.cfName,
                                                  indexMetadata.name));
                        return type;
                    }
                }

                warn.accept(String.format("sstable %s: contains column '%s', which is not present in the schema",
                                          desc,
                                          logColumnName(name)));
            }
            else
            {
                // This is a best-effort approach to handle the case of a UDT column created *AND* dropped in
                // C* 3.0.
                if (type instanceof UserType && cd.type instanceof TupleType)
                {
                    // At this point, we know that the type belongs to a dropped column, recorded with the
                    // dropped column type "TupleType" and using "UserType" in the sstable. So it is very
                    // likely, that this belongs to a dropped UDT. Fix that information to tuple-type.
                    return fixType(desc, name, type, cd.type, true);
                }
            }

            return type;
        }

        // At this point, the column name is known to be a "non-dropped" column in the table.
        if (cd.kind != kind)
            error("sstable %s: contains column '%s' as a %s column, but is of kind %s in the schema",
                  desc,
                  logColumnName(name),
                  kind.name().toLowerCase(),
                  cd.kind.name().toLowerCase());
        else
            type = fixType(desc, name, type, cd.type, false);
        return type;
    }

    private AbstractType<?> fixType(Descriptor desc, ByteBuffer name, AbstractType<?> typeInHeader, AbstractType<?> typeInSchema, boolean droppedColumnMode)
    {
        AbstractType<?> fixedType = fixTypeInner(typeInHeader, typeInSchema, droppedColumnMode);
        if (fixedType != null)
        {
            if (fixedType != typeInHeader)
                info.accept(String.format("sstable %s: Column '%s' needs to be updated from type '%s' to '%s'",
                                          desc,
                                          logColumnName(name),
                                          typeInHeader.asCQL3Type(),
                                          fixedType.asCQL3Type()));
            return fixedType;
        }

        error("sstable %s: contains column '%s' as type '%s', but schema mentions '%s'",
              desc,
              logColumnName(name),
              typeInHeader.asCQL3Type(),
              typeInSchema.asCQL3Type());

        return typeInHeader;
    }

    private AbstractType<?> fixTypeInner(AbstractType<?> typeInHeader, AbstractType<?> typeInSchema, boolean droppedColumnMode)
    {
        if (typeEquals(typeInHeader, typeInSchema))
            return typeInHeader;

        if (typeInHeader instanceof CollectionType)
            return fixTypeInnerCollection(typeInHeader, typeInSchema, droppedColumnMode);

        if (typeInHeader instanceof AbstractCompositeType)
            return fixTypeInnerAbstractComposite(typeInHeader, typeInSchema, droppedColumnMode);

        if (typeInHeader instanceof TupleType)
            return fixTypeInnerAbstractTuple(typeInHeader, typeInSchema, droppedColumnMode);

        // all types, beside CollectionType + AbstractCompositeType + TupleType, should be ok (no nested types) - just check for compatibility
        if (typeInHeader.isCompatibleWith(typeInSchema))
            return typeInHeader;

        return null;
    }

    private AbstractType<?> fixTypeInnerAbstractTuple(AbstractType<?> typeInHeader, AbstractType<?> typeInSchema, boolean droppedColumnMode)
    {
        // This first 'if' handles the case when a UDT has been dropped, as a dropped UDT is recorded as a tuple
        // in dropped_columns. If a UDT is to be replaced with a tuple, then also do that for the inner UDTs.
        if (droppedColumnMode && typeInHeader.getClass() == UserType.class && typeInSchema instanceof TupleType)
            return fixTypeInnerUserTypeDropped((UserType) typeInHeader, (TupleType) typeInSchema);

        if (typeInHeader.getClass() != typeInSchema.getClass())
            return null;

        if (typeInHeader.getClass() == UserType.class)
            return fixTypeInnerUserType((UserType) typeInHeader, (UserType) typeInSchema);

        if (typeInHeader.getClass() == TupleType.class)
            return fixTypeInnerTuple((TupleType) typeInHeader, (TupleType) typeInSchema, droppedColumnMode);

        throw new IllegalArgumentException("Unknown tuple type class " + typeInHeader.getClass().getName());
    }

    private AbstractType<?> fixTypeInnerCollection(AbstractType<?> typeInHeader, AbstractType<?> typeInSchema, boolean droppedColumnMode)
    {
        if (typeInHeader.getClass() != typeInSchema.getClass())
            return null;

        if (typeInHeader.getClass() == ListType.class)
            return fixTypeInnerList((ListType<?>) typeInHeader, (ListType<?>) typeInSchema, droppedColumnMode);

        if (typeInHeader.getClass() == SetType.class)
            return fixTypeInnerSet((SetType<?>) typeInHeader, (SetType<?>) typeInSchema, droppedColumnMode);

        if (typeInHeader.getClass() == MapType.class)
            return fixTypeInnerMap((MapType<?, ?>) typeInHeader, (MapType<?, ?>) typeInSchema, droppedColumnMode);

        throw new IllegalArgumentException("Unknown collection type class " + typeInHeader.getClass().getName());
    }

    private AbstractType<?> fixTypeInnerAbstractComposite(AbstractType<?> typeInHeader, AbstractType<?> typeInSchema, boolean droppedColumnMode)
    {
        if (typeInHeader.getClass() != typeInSchema.getClass())
            return null;

        if (typeInHeader.getClass() == CompositeType.class)
            return fixTypeInnerComposite((CompositeType) typeInHeader, (CompositeType) typeInSchema, droppedColumnMode);

        if (typeInHeader.getClass() == DynamicCompositeType.class)
        {
            // Not sure if we should care about UDTs in DynamicCompositeType at all...
            if (!typeInHeader.isCompatibleWith(typeInSchema))
                return null;

            return typeInHeader;
        }

        throw new IllegalArgumentException("Unknown composite type class " + typeInHeader.getClass().getName());
    }

    private AbstractType<?> fixTypeInnerUserType(UserType cHeader, UserType cSchema)
    {
        if (!cHeader.keyspace.equals(cSchema.keyspace) || !cHeader.name.equals(cSchema.name))
            // different UDT - bummer...
            return null;

        if (cHeader.isMultiCell() != cSchema.isMultiCell())
        {
            if (cHeader.isMultiCell() && !cSchema.isMultiCell())
            {
                // C* 3.0 writes broken SerializationHeader.Component instances - i.e. broken UDT type
                // definitions into the sstable -Stats.db file, because 3.0 does not enclose frozen UDTs
                // (and all UDTs in 3.0 were frozen) with an '' bracket. Since CASSANDRA-7423 (support
                // for non-frozen UDTs, committed to C* 3.6), that frozen-bracket is quite important.
                // Non-frozen (= multi-cell) UDTs are serialized in a fundamentally different way than
                // frozen UDTs in sstables - most importantly, the order of serialized columns depends on
                // the type: fixed-width types first, then variable length types (like frozen types),
                // multi-cell types last. If C* >= 3.6 reads an sstable with a UDT that's written by
                // C* < 3.6, a variety of CorruptSSTableExceptions get logged and clients will encounter
                // read errors.
                // At this point, we know that the type belongs to a "live" (non-dropped) column, so it
                // is safe to correct the information from the header.
                return cSchema;
            }

            // In all other cases, there's not much we can do.
            return null;
        }

        return cHeader;
    }

    private AbstractType<?> fixTypeInnerUserTypeDropped(UserType cHeader, TupleType cSchema)
    {
        // Do not mess around with the UserType in the serialization header, if the column has been dropped.
        // Only fix the multi-cell status when the header contains it as a multicell (non-frozen) UserType,
        // but the schema says "frozen".
        if (cHeader.isMultiCell() && !cSchema.isMultiCell())
        {
            return new UserType(cHeader.keyspace, cHeader.name, cHeader.fieldNames(), cHeader.fieldTypes(), cSchema.isMultiCell());
        }

        return cHeader;
    }

    private AbstractType<?> fixTypeInnerTuple(TupleType cHeader, TupleType cSchema, boolean droppedColumnMode)
    {
        if (cHeader.size() != cSchema.size())
            // different number of components - bummer...
            return null;
        List<AbstractType<?>> cHeaderFixed = new ArrayList<>(cHeader.size());
        boolean anyChanged = false;
        for (int i = 0; i < cHeader.size(); i++)
        {
            AbstractType<?> cHeaderComp = cHeader.type(i);
            AbstractType<?> cHeaderCompFixed = fixTypeInner(cHeaderComp, cSchema.type(i), droppedColumnMode);
            if (cHeaderCompFixed == null)
                // incompatible, bummer...
                return null;
            cHeaderFixed.add(cHeaderCompFixed);
            anyChanged |= cHeaderComp != cHeaderCompFixed;
        }
        if (anyChanged || cSchema.isMultiCell() != cHeader.isMultiCell())
            // TODO this should create a non-frozen tuple type for the sake of handling a dropped, non-frozen UDT
            return new TupleType(cHeaderFixed);
        return cHeader;
    }

    private AbstractType<?> fixTypeInnerComposite(CompositeType cHeader, CompositeType cSchema, boolean droppedColumnMode)
    {
        if (cHeader.types.size() != cSchema.types.size())
            // different number of components - bummer...
            return null;
        List<AbstractType<?>> cHeaderFixed = new ArrayList<>(cHeader.types.size());
        boolean anyChanged = false;
        for (int i = 0; i < cHeader.types.size(); i++)
        {
            AbstractType<?> cHeaderComp = cHeader.types.get(i);
            AbstractType<?> cHeaderCompFixed = fixTypeInner(cHeaderComp, cSchema.types.get(i), droppedColumnMode);
            if (cHeaderCompFixed == null)
                // incompatible, bummer...
                return null;
            cHeaderFixed.add(cHeaderCompFixed);
            anyChanged |= cHeaderComp != cHeaderCompFixed;
        }
        if (anyChanged)
            return CompositeType.getInstance(cHeaderFixed);
        return cHeader;
    }

    private AbstractType<?> fixTypeInnerList(ListType<?> cHeader, ListType<?> cSchema, boolean droppedColumnMode)
    {
        AbstractType<?> cHeaderElem = cHeader.getElementsType();
        AbstractType<?> cHeaderElemFixed = fixTypeInner(cHeaderElem, cSchema.getElementsType(), droppedColumnMode);
        if (cHeaderElemFixed == null)
            // bummer...
            return null;
        if (cHeaderElem != cHeaderElemFixed)
            // element type changed
            return ListType.getInstance(cHeaderElemFixed, cHeader.isMultiCell());
        return cHeader;
    }

    private AbstractType<?> fixTypeInnerSet(SetType<?> cHeader, SetType<?> cSchema, boolean droppedColumnMode)
    {
        AbstractType<?> cHeaderElem = cHeader.getElementsType();
        AbstractType<?> cHeaderElemFixed = fixTypeInner(cHeaderElem, cSchema.getElementsType(), droppedColumnMode);
        if (cHeaderElemFixed == null)
            // bummer...
            return null;
        if (cHeaderElem != cHeaderElemFixed)
            // element type changed
            return SetType.getInstance(cHeaderElemFixed, cHeader.isMultiCell());
        return cHeader;
    }

    private AbstractType<?> fixTypeInnerMap(MapType<?, ?> cHeader, MapType<?, ?> cSchema, boolean droppedColumnMode)
    {
        AbstractType<?> cHeaderKey = cHeader.getKeysType();
        AbstractType<?> cHeaderVal = cHeader.getValuesType();
        AbstractType<?> cHeaderKeyFixed = fixTypeInner(cHeaderKey, cSchema.getKeysType(), droppedColumnMode);
        AbstractType<?> cHeaderValFixed = fixTypeInner(cHeaderVal, cSchema.getValuesType(), droppedColumnMode);
        if (cHeaderKeyFixed == null || cHeaderValFixed == null)
            // bummer...
            return null;
        if (cHeaderKey != cHeaderKeyFixed || cHeaderVal != cHeaderValFixed)
            // element type changed
            return MapType.getInstance(cHeaderKeyFixed, cHeaderValFixed, cHeader.isMultiCell());
        return cHeader;
    }

    private boolean typeEquals(AbstractType<?> typeInHeader, AbstractType<?> typeInSchema)
    {
        // Quite annoying, but the implementations of equals() on some implementation of AbstractType seems to be
        // wrong, but toString() seems to work in such cases.
        return typeInHeader.equals(typeInSchema) || typeInHeader.toString().equals(typeInSchema.toString());
    }

    private static String logColumnName(ByteBuffer columnName)
    {
        try
        {
            return ByteBufferUtil.string(columnName);
        }
        catch (CharacterCodingException e)
        {
            return "?? " + e;
        }
    }

    private Map<MetadataType, MetadataComponent> readSSTableMetadata(Descriptor desc)
    {
        Map<MetadataType, MetadataComponent> metadata;
        try
        {
            metadata = desc.getMetadataSerializer().deserialize(desc, EnumSet.allOf(MetadataType.class));
        }
        catch (IOException e)
        {
            error("Failed to deserialize metadata for sstable %s: %s", desc, e.toString());
            return null;
        }
        return metadata;
    }

    private void writeNewMetadata(Descriptor desc, Map<MetadataType, MetadataComponent> newMetadata)
    {
        String file = desc.filenameFor(Component.STATS);
        info.accept(String.format("  Writing new metadata file %s", file));
        try
        {
            desc.getMetadataSerializer().rewriteSSTableMetadata(desc, newMetadata);
        }
        catch (IOException e)
        {
            error("Failed to write metadata component for %s: %s", file, e.toString());
            throw new RuntimeException(e);
        }
    }

    /**
     * Fix individually provided sstables or directories containing sstables.
     */
    static class ManualHeaderFix extends SSTableHeaderFix
    {
        private final List<Path> paths;

        ManualHeaderFix(Builder builder)
        {
            super(builder);
            this.paths = builder.paths;
        }

        public void prepare()
        {
            paths.forEach(this::processFileOrDirectory);
        }
    }

    /**
     * Fix all sstables in the configured data-directories.
     */
    static class AutomaticHeaderFix extends SSTableHeaderFix
    {
        AutomaticHeaderFix(Builder builder)
        {
            super(builder);
        }

        public void prepare()
        {
            info.accept("Scanning all data directories...");
            for (Directories.DataDirectory dataDirectory : Directories.dataDirectories)
                scanDataDirectory(dataDirectory);
            info.accept("Finished scanning all data directories...");
        }

        private void scanDataDirectory(Directories.DataDirectory dataDirectory)
        {
            info.accept(String.format("Scanning data directory %s", dataDirectory.location));
            File[] ksDirs = dataDirectory.location.listFiles();
            if (ksDirs == null)
                return;
            for (File ksDir : ksDirs)
            {
                if (!ksDir.isDirectory() || !ksDir.canRead())
                    continue;

                String name = ksDir.getName();

                // silently ignore all system keyspaces
                if (SchemaConstants.isLocalSystemKeyspace(name) || SchemaConstants.isReplicatedSystemKeyspace(name))
                    continue;

                File[] tabDirs = ksDir.listFiles();
                if (tabDirs == null)
                    continue;
                for (File tabDir : tabDirs)
                {
                    if (!tabDir.isDirectory() || !tabDir.canRead())
                        continue;

                    processFileOrDirectory(tabDir.toPath());
                }
            }
        }
    }
}