DataComponent.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.cassandra.io.sstable.format;

import org.apache.cassandra.config.Config.FlushCompression;
import org.apache.cassandra.db.compaction.OperationType;
import org.apache.cassandra.io.compress.CompressedSequentialWriter;
import org.apache.cassandra.io.compress.ICompressor;
import org.apache.cassandra.io.sstable.Descriptor;
import org.apache.cassandra.io.sstable.format.SSTableFormat.Components;
import org.apache.cassandra.io.sstable.metadata.MetadataCollector;
import org.apache.cassandra.io.util.ChecksummedSequentialWriter;
import org.apache.cassandra.io.util.SequentialWriter;
import org.apache.cassandra.io.util.SequentialWriterOption;
import org.apache.cassandra.schema.CompressionParams;
import org.apache.cassandra.schema.TableMetadata;

public class DataComponent
{
    public static SequentialWriter buildWriter(Descriptor descriptor,
                                               TableMetadata metadata,
                                               SequentialWriterOption options,
                                               MetadataCollector metadataCollector,
                                               OperationType operationType,
                                               FlushCompression flushCompression)
    {
        if (metadata.params.compression.isEnabled())
        {
            final CompressionParams compressionParams = buildCompressionParams(metadata, operationType, flushCompression);

            return new CompressedSequentialWriter(descriptor.fileFor(Components.DATA),
                                                  descriptor.fileFor(Components.COMPRESSION_INFO),
                                                  descriptor.fileFor(Components.DIGEST),
                                                  options,
                                                  compressionParams,
                                                  metadataCollector);
        }
        else
        {
            return new ChecksummedSequentialWriter(descriptor.fileFor(Components.DATA),
                                                   descriptor.fileFor(Components.CRC),
                                                   descriptor.fileFor(Components.DIGEST),
                                                   options);
        }
    }

    /**
     * Given an OpType, determine the correct Compression Parameters
     *
     * @return {@link CompressionParams}
     */
    private static CompressionParams buildCompressionParams(TableMetadata metadata, OperationType operationType, FlushCompression flushCompression)
    {
        CompressionParams compressionParams = metadata.params.compression;
        final ICompressor compressor = compressionParams.getSstableCompressor();

        if (null != compressor && operationType == OperationType.FLUSH)
        {
            // When we are flushing out of the memtable throughput of the compressor is critical as flushes,
            // especially of large tables, can queue up and potentially block writes.
            // This optimization allows us to fall back to a faster compressor if a particular
            // compression algorithm indicates we should. See CASSANDRA-15379 for more details.
            switch (flushCompression)
            {
                // It is relatively easier to insert a Noop compressor than to disable compressed writing
                // entirely as the "compression" member field is provided outside the scope of this class.
                // It may make sense in the future to refactor the ownership of the compression flag so that
                // We can bypass the CompressedSequentialWriter in this case entirely.
                case none:
                    compressionParams = CompressionParams.NOOP;
                    break;
                case fast:
                    if (!compressor.recommendedUses().contains(ICompressor.Uses.FAST_COMPRESSION))
                    {
                        // The default compressor is generally fast (LZ4 with 16KiB block size)
                        compressionParams = CompressionParams.DEFAULT;
                        break;
                    }
                    // else fall through
                case table:
                default:
                    break;
            }
        }
        return compressionParams;
    }
}