test/unit/org/apache/cassandra/io/sstable/SSTableFormatTest.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;

import java.io.IOException;
import java.net.URL;
import java.util.Arrays;
import java.util.Map;
import java.util.function.BiFunction;

import com.google.common.collect.ImmutableMap;
import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.Test;

import org.apache.cassandra.config.Config;
import org.apache.cassandra.config.Config.SSTableConfig;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.config.YamlConfigurationLoader;
import org.apache.cassandra.exceptions.ConfigurationException;
import org.apache.cassandra.io.sstable.format.SSTableFormat;
import org.apache.cassandra.io.sstable.format.SSTableReader;
import org.apache.cassandra.io.sstable.format.SSTableWriter;
import org.apache.cassandra.io.sstable.format.Version;
import org.apache.cassandra.io.util.File;
import org.apache.cassandra.io.util.FileOutputStreamPlus;
import org.apache.cassandra.io.util.FileUtils;
import org.mockito.Mockito;

import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
import static org.mockito.Mockito.when;

public class SSTableFormatTest
{
    public static abstract class AbstractFormat implements SSTableFormat<SSTableReader, SSTableWriter>
    {
        public Map<String, String> options;
        public String name;
        public String latestVersion;

        public AbstractFormat(String latestVersion)
        {
            this.latestVersion = latestVersion;
        }

        @Override
        public Version getVersion(String version)
        {
            Version v = Mockito.mock(Version.class);
            when(v.toString()).thenReturn(version);
            when(v.isCompatible()).thenReturn(version.charAt(0) == latestVersion.charAt(0));
            return v;
        }

        @Override
        public Version getLatestVersion()
        {
            return getVersion(latestVersion);
        }

        @Override
        public String name()
        {
            return name;
        }

        static class Factory implements SSTableFormat.Factory
        {
            public String name;
            public BiFunction<Map<String, String>, String, SSTableFormat<?, ?>> provider;

            public Factory(String name, BiFunction<Map<String, String>, String, SSTableFormat<?, ?>> provider)
            {
                this.name = name;
                this.provider = provider;
            }

            @Override
            public String name()
            {
                return name;
            }

            @Override
            public SSTableFormat<?, ?> getInstance(Map<String, String> options)
            {
                return provider.apply(options, name);
            }
        }
    }

    public static AbstractFormat.Factory factory(String name, Class<? extends AbstractFormat> clazz)
    {
        return new AbstractFormat.Factory(name, (options, version) -> {
            AbstractFormat format = Mockito.spy(clazz);
            format.name = name;
            format.options = options;
            return format;
        });
    }

    public static abstract class Format1 extends AbstractFormat
    {
        public Format1()
        {
            super("xx");
        }
    }

    public static abstract class Format2 extends AbstractFormat
    {
        public Format2()
        {
            super("yy");
        }
    }

    public static abstract class Format3 extends AbstractFormat
    {
        public Format3()
        {
            super("zz");
        }
    }

    @BeforeClass
    public static void beforeClass()
    {
        DatabaseDescriptor.clientInitialization();
    }

    private static final String yamlContent0 = "";
    private static final SSTableConfig expected0 = new Config.SSTableConfig();

    private static final String yamlContent1 = "sstable:\n" +
                                               "   selected_format: aaa\n";
    private static final SSTableConfig expected1 = new Config.SSTableConfig()
    {
        {
            selected_format = "aaa";
        }
    };

    private static final String yamlContent2 = "sstable:\n" +
                                               "   selected_format: aaa\n" +
                                               "   format:\n" +
                                               "       aaa:\n" +
                                               "           param1: value1\n" +
                                               "           param2: value2\n" +
                                               "       bbb:\n" +
                                               "           param3: value3\n" +
                                               "           param4: value4\n";

    private static final Config.SSTableConfig expected2 = new SSTableConfig()
    {
        {
            selected_format = "aaa";
            format = ImmutableMap.of("aaa", ImmutableMap.of("param1", "value1", "param2", "value2"),
                                     "bbb", ImmutableMap.of("param3", "value3", "param4", "value4"));
        }
    };

    private static final SSTableConfig unexpected = new Config.SSTableConfig()
    {
        {
            selected_format = "aaa";
        }
    };


    @Test
    public void testParsingYamlConfig() throws IOException
    {
        YamlConfigurationLoader loader = new YamlConfigurationLoader();
        File f = FileUtils.createTempFile("sstable_format_test_config", ".yaml");
        URL url = f.toPath().toUri().toURL();

        ImmutableMap.of(yamlContent0, expected0, yamlContent1, expected1, yamlContent2, expected2).forEach((yamlContent, expected) -> {
            try (FileOutputStreamPlus out = f.newOutputStream(File.WriteMode.OVERWRITE))
            {
                out.write(yamlContent.getBytes());
            }
            catch (IOException e)
            {
                throw new RuntimeException(e);
            }
            Config config = loader.loadConfig(url);
            assertThat(config.sstable).describedAs("Yaml: \n%s\n", yamlContent).isEqualToComparingFieldByField(expected);
        });
    }

    @Before
    public void before()
    {
    }

    public static void configure(SSTableConfig config, SSTableFormat.Factory... factories)
    {
        DatabaseDescriptor.resetSSTableFormats(Arrays.asList(factories), config);
    }

    private void verifyFormat(String name, Map<String, String> options)
    {
        AbstractFormat format = (AbstractFormat) DatabaseDescriptor.getSSTableFormats().get(name);
        assertThat(format.name).isEqualTo(name);
        assertThat(format.options).isEqualTo(options);
    }

    private void verifySelectedFormat(String name)
    {
        assertThat(DatabaseDescriptor.getSelectedSSTableFormat().name()).isEqualTo(name);
    }

    @Test
    public void testValidConfig()
    {
        configure(expected1, factory("aaa", Format1.class));
        assertThat(DatabaseDescriptor.getSSTableFormats()).hasSize(1);
        verifyFormat("aaa", ImmutableMap.of());
        verifySelectedFormat("aaa");

        configure(expected2, factory("aaa", Format1.class), factory("bbb", Format2.class), factory("ccc", Format3.class));
        assertThat(DatabaseDescriptor.getSSTableFormats()).hasSize(3);
        verifyFormat("aaa", ImmutableMap.of("param1", "value1", "param2", "value2"));
        verifyFormat("bbb", ImmutableMap.of("param3", "value3", "param4", "value4"));
        verifyFormat("ccc", ImmutableMap.of());
        verifySelectedFormat("aaa");
    }

    @Test
    public void testConfigValidation()
    {
        // invalid name
        assertThatExceptionOfType(ConfigurationException.class).isThrownBy(() -> configure(expected1, factory("Aa", Format1.class)))
                                                               .withMessageContainingAll("SSTable format name", "must be non-empty, lower-case letters only string");
        assertThatExceptionOfType(ConfigurationException.class).isThrownBy(() -> configure(expected1, factory("a-a", Format1.class)))
                                                               .withMessageContainingAll("SSTable format name", "must be non-empty, lower-case letters only string");
        assertThatExceptionOfType(ConfigurationException.class).isThrownBy(() -> configure(expected1, factory("a1", Format1.class)))
                                                               .withMessageContainingAll("SSTable format name", "must be non-empty, lower-case letters only string");
        assertThatExceptionOfType(ConfigurationException.class).isThrownBy(() -> configure(expected1, factory("", Format1.class)))
                                                               .withMessageContainingAll("SSTable format name", "must be non-empty, lower-case letters only string");

        // duplicate name
        assertThatExceptionOfType(ConfigurationException.class).isThrownBy(() -> configure(expected1, factory("aaa", Format1.class), factory("aaa", Format2.class)))
                                                               .withMessageContainingAll("Multiple sstable format implementations with the same name", "aaa");

        // missing name
        assertThatExceptionOfType(ConfigurationException.class).isThrownBy(() -> configure(expected1, factory(null, Format1.class)))
                                                               .withMessageContainingAll("SSTable format name", "cannot be null");

        // Configuration contains options of unknown sstable formats
        assertThatExceptionOfType(ConfigurationException.class).isThrownBy(() -> configure(expected2, factory("aaa", Format1.class)))
                                                               .withMessageContainingAll("Configuration contains options of unknown sstable formats", "bbb");

        // Selected sstable format '%s' is not available
        assertThatExceptionOfType(ConfigurationException.class).isThrownBy(() -> configure(expected1, factory("bbb", Format1.class)))
                                                               .withMessageContainingAll("Selected sstable format", "aaa", "is not available");
    }
}