/*
 * 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.db.filter;

import java.io.IOException;
import java.util.function.Consumer;

import com.google.common.base.Throwables;
import org.junit.Assert;
import org.junit.Test;

import org.apache.cassandra.config.CFMetaData;
import org.apache.cassandra.config.ColumnDefinition;
import org.apache.cassandra.cql3.ColumnIdentifier;
import org.apache.cassandra.db.PartitionColumns;
import org.apache.cassandra.db.marshal.Int32Type;
import org.apache.cassandra.db.marshal.SetType;
import org.apache.cassandra.db.rows.CellPath;
import org.apache.cassandra.dht.Murmur3Partitioner;
import org.apache.cassandra.io.util.DataInputBuffer;
import org.apache.cassandra.io.util.DataInputPlus;
import org.apache.cassandra.io.util.DataOutputBuffer;
import org.apache.cassandra.net.MessagingService;
import org.apache.cassandra.utils.ByteBufferUtil;

import static org.junit.Assert.assertEquals;

public class ColumnFilterTest
{
    private static final ColumnFilter.Serializer serializer = new ColumnFilter.Serializer();

    private final CFMetaData metadata = CFMetaData.Builder.create("ks", "table")
                                                          .withPartitioner(Murmur3Partitioner.instance)
                                                          .addPartitionKey("pk", Int32Type.instance)
                                                          .addClusteringColumn("ck", Int32Type.instance)
                                                          .addStaticColumn("s1", Int32Type.instance)
                                                          .addStaticColumn("s2", SetType.getInstance(Int32Type.instance, true))
                                                          .addRegularColumn("v1", Int32Type.instance)
                                                          .addRegularColumn("v2", SetType.getInstance(Int32Type.instance, true))
                                                          .addRegularColumn(ColumnIdentifier.getInterned("Escaped Name", true), Int32Type.instance)
                                                          .build();

    private final ColumnDefinition s1 = metadata.getColumnDefinition(ByteBufferUtil.bytes("s1"));
    private final ColumnDefinition s2 = metadata.getColumnDefinition(ByteBufferUtil.bytes("s2"));
    private final ColumnDefinition v1 = metadata.getColumnDefinition(ByteBufferUtil.bytes("v1"));
    private final ColumnDefinition v2 = metadata.getColumnDefinition(ByteBufferUtil.bytes("v2"));
    private final ColumnDefinition escaped = metadata.getColumnDefinition(ByteBufferUtil.bytes("Escaped Name"));
    private final CellPath path0 = CellPath.create(ByteBufferUtil.bytes(0));
    private final CellPath path1 = CellPath.create(ByteBufferUtil.bytes(1));
    private final CellPath path2 = CellPath.create(ByteBufferUtil.bytes(2));
    private final CellPath path3 = CellPath.create(ByteBufferUtil.bytes(3));
    private final CellPath path4 = CellPath.create(ByteBufferUtil.bytes(4));

    // Select all

    @Test
    public void testSelectAll()
    {
        Consumer<ColumnFilter> check = filter -> {
            testRoundTrips(filter);
            assertEquals("*/*", filter.toString());
            assertEquals("*", filter.toCQLString());
            assertFetchedQueried(true, true, filter, v1, v2, s1, s2);
            assertCellFetchedQueried(true, true, filter, v2, path0, path1, path2, path3, path4);
            assertCellFetchedQueried(true, true, filter, s2, path0, path1, path2, path3, path4);
        };

        check.accept(ColumnFilter.all(metadata));
        check.accept(ColumnFilter.allColumnsBuilder(metadata).build());
    }

    // Selections

    @Test
    public void testSelectNothing()
    {
        Consumer<ColumnFilter> check = filter -> {
            testRoundTrips(filter);
            assertEquals("[]", filter.toString());
            assertEquals("*", filter.toCQLString());
            assertFetchedQueried(false, false, filter, v1, v2, s1, s2);
            assertCellFetchedQueried(false, false, filter, v2, path0, path1, path2, path3, path4);
            assertCellFetchedQueried(false, false, filter, s2, path0, path1, path2, path3, path4);
        };

        check.accept(ColumnFilter.selection(PartitionColumns.NONE));
        check.accept(ColumnFilter.selectionBuilder().build());
    }

    @Test
    public void testSelectSimpleColumn()
    {
        Consumer<ColumnFilter> check = filter -> {
            testRoundTrips(filter);
            assertEquals("[v1]", filter.toString());
            assertEquals("v1", filter.toCQLString());
            assertFetchedQueried(true, true, filter, v1);
            assertFetchedQueried(false, false, filter, v2, s1, s2);
            assertCellFetchedQueried(false, false, filter, v2, path0, path1, path2, path3, path4);
            assertCellFetchedQueried(false, false, filter, s2, path0, path1, path2, path3, path4);
        };

        check.accept(ColumnFilter.selection(PartitionColumns.builder().add(v1).build()));
        check.accept(ColumnFilter.selectionBuilder().add(v1).build());
    }

    @Test
    public void testSelectEscapedColumn()
    {
        Consumer<ColumnFilter> check = filter -> {
            testRoundTrips(filter);
            assertEquals("[Escaped Name]", filter.toString());
            assertEquals("\"Escaped Name\"", filter.toCQLString());
            assertFetchedQueried(true, true, filter, escaped);
            assertFetchedQueried(false, false, filter, v2, s1, s2);
            assertCellFetchedQueried(false, false, filter, v2, path0, path1, path2, path3, path4);
            assertCellFetchedQueried(false, false, filter, s2, path0, path1, path2, path3, path4);
        };

        check.accept(ColumnFilter.selection(PartitionColumns.builder().add(escaped).build()));
        check.accept(ColumnFilter.selectionBuilder().add(escaped).build());
    }

    @Test
    public void testSelectComplexColumn()
    {
        Consumer<ColumnFilter> check = filter -> {
            testRoundTrips(filter);
            assertEquals("[v2]", filter.toString());
            assertEquals("v2", filter.toCQLString());
            assertFetchedQueried(true, true, filter, v2);
            assertFetchedQueried(false, false, filter, v1, s1, s2);
            assertCellFetchedQueried(true, true, filter, v2, path0, path1, path2, path3, path4);
            assertCellFetchedQueried(false, false, filter, s2, path0, path1, path2, path3, path4);
        };

        check.accept(ColumnFilter.selection(PartitionColumns.builder().add(v2).build()));
        check.accept(ColumnFilter.selectionBuilder().add(v2).build());
    }

    @Test
    public void testSelectStaticColumn()
    {
        Consumer<ColumnFilter> check = filter -> {
            testRoundTrips(filter);
            assertEquals("[s1]", filter.toString());
            assertEquals("s1", filter.toCQLString());
            assertFetchedQueried(true, true, filter, s1);
            assertFetchedQueried(false, false, filter, v1, v2, s2);
            assertCellFetchedQueried(false, false, filter, v2, path0, path1, path2, path3, path4);
            assertCellFetchedQueried(false, false, filter, s2, path0, path1, path2, path3, path4);
        };

        check.accept(ColumnFilter.selection(PartitionColumns.builder().add(s1).build()));
        check.accept(ColumnFilter.selectionBuilder().add(s1).build());
    }

    @Test
    public void testSelectStaticComplexColumn()
    {
        Consumer<ColumnFilter> check = filter -> {
            testRoundTrips(filter);
            assertEquals("[s2]", filter.toString());
            assertEquals("s2", filter.toCQLString());
            assertFetchedQueried(true, true, filter, s2);
            assertFetchedQueried(false, false, filter, v1, v2, s1);
            assertCellFetchedQueried(false, false, filter, v2, path0, path1, path2, path3, path4);
            assertCellFetchedQueried(true, true, filter, s2, path0, path1, path2, path3, path4);
        };

        check.accept(ColumnFilter.selection(PartitionColumns.builder().add(s2).build()));
        check.accept(ColumnFilter.selectionBuilder().add(s2).build());
    }

    @Test
    public void testSelectColumns()
    {
        Consumer<ColumnFilter> check = filter -> {
            testRoundTrips(filter);
            assertEquals("[s1, s2, v1, v2]", filter.toString());
            assertEquals("s1, s2, v1, v2", filter.toCQLString());
            assertFetchedQueried(true, true, filter, v1, v2, s1, s2);
            assertCellFetchedQueried(true, true, filter, v2, path0, path1, path2, path3, path4);
            assertCellFetchedQueried(true, true, filter, s2, path0, path1, path2, path3, path4);
        };

        check.accept(ColumnFilter.selection(PartitionColumns.builder().add(v1).add(v2).add(s1).add(s2).build()));
        check.accept(ColumnFilter.selectionBuilder().add(v1).add(v2).add(s1).add(s2).build());
    }

    @Test
    public void testSelectIndividualCells()
    {
        ColumnFilter filter = ColumnFilter.selectionBuilder().select(v2, path1).select(v2, path3).build();
        testRoundTrips(filter);
        assertEquals("[v2[1], v2[3]]", filter.toString());
        assertEquals("v2[1], v2[3]", filter.toCQLString());
        assertFetchedQueried(true, true, filter, v2);
        assertFetchedQueried(false, false, filter, v1, s1, s2);
        assertCellFetchedQueried(true, true, filter, v2, path1, path3);
        assertCellFetchedQueried(false, false, filter, v2, path0, path2, path4);
        assertCellFetchedQueried(false, false, filter, s2, path0, path1, path2, path3, path4);
    }

    @Test
    public void testSelectIndividualCellsFromStatic()
    {
        ColumnFilter filter = ColumnFilter.selectionBuilder().select(s2, path1).select(s2, path3).build();
        testRoundTrips(filter);
        assertEquals("[s2[1], s2[3]]", filter.toString());
        assertEquals("s2[1], s2[3]", filter.toCQLString());
        assertFetchedQueried(true, true, filter, s2);
        assertFetchedQueried(false, false, filter, v1, v2, s1);
        assertCellFetchedQueried(false, false, filter, v2, path0, path1, path2, path3, path4);
        assertCellFetchedQueried(true, true, filter, s2, path1, path3);
        assertCellFetchedQueried(false, false, filter, s2, path0, path2, path4);
    }

    @Test
    public void testSelectCellSlice()
    {
        ColumnFilter filter = ColumnFilter.selectionBuilder().slice(v2, path1, path3).build();
        testRoundTrips(filter);
        assertEquals("[v2[1:3]]", filter.toString());
        assertEquals("v2[1:3]", filter.toCQLString());
        assertFetchedQueried(true, true, filter, v2);
        assertFetchedQueried(false, false, filter, v1, s1, s2);
        assertCellFetchedQueried(true, true, filter, v2, path1, path2, path3);
        assertCellFetchedQueried(false, false, filter, v2, path0, path4);
        assertCellFetchedQueried(false, false, filter, s2, path0, path1, path2, path3, path4);
    }

    @Test
    public void testSelectCellSliceFromStatic()
    {
        ColumnFilter filter = ColumnFilter.selectionBuilder().slice(s2, path1, path3).build();
        testRoundTrips(filter);
        assertEquals("[s2[1:3]]", filter.toString());
        assertEquals("s2[1:3]", filter.toCQLString());
        assertFetchedQueried(true, true, filter, s2);
        assertFetchedQueried(false, false, filter, v1, v2, s1);
        assertCellFetchedQueried(false, false, filter, v2, path0, path1, path2, path3, path4);
        assertCellFetchedQueried(true, true, filter, s2, path1, path2, path3);
        assertCellFetchedQueried(false, false, filter, s2, path0, path4);
    }

    @Test
    public void testSelectColumnsWithCellsAndSlices()
    {
        ColumnFilter filter = ColumnFilter.selectionBuilder()
                                          .add(v1)
                                          .add(s1)
                                          .slice(v2, path0, path2)
                                          .select(v2, path4)
                                          .select(s2, path0)
                                          .slice(s2, path2, path4)
                                          .build();
        testRoundTrips(filter);
        assertEquals("[s1, s2[0], s2[2:4], v1, v2[0:2], v2[4]]", filter.toString());
        assertEquals("s1, s2[0], s2[2:4], v1, v2[0:2], v2[4]", filter.toCQLString());
        assertFetchedQueried(true, true, filter, v1, v2, s1, s2);
        assertCellFetchedQueried(true, true, filter, v2, path0, path1, path2, path4);
        assertCellFetchedQueried(false, false, filter, v2, path3);
        assertCellFetchedQueried(true, true, filter, s2, path0, path2, path3, path4);
        assertCellFetchedQueried(false, false, filter, s2, path1);
    }

    // select with metadata

    @Test
    public void testSelectSimpleColumnWithMetadata()
    {
        Consumer<ColumnFilter> check = filter -> {
            testRoundTrips(filter);
            assertFetchedQueried(true, true, filter, v1);

            assertEquals("*/*", filter.toString());
            assertEquals("v1", filter.toCQLString());
            assertFetchedQueried(true, true, filter, s1, s2, v2);
            assertCellFetchedQueried(true, true, filter, v2, path0, path1, path2, path3, path4);
            assertCellFetchedQueried(true, true, filter, s2, path0, path1, path2, path3, path4);
        };

        check.accept(ColumnFilter.selection(metadata, PartitionColumns.builder().add(v1).build()));
        check.accept(ColumnFilter.allColumnsBuilder(metadata).add(v1).build());
    }

    @Test
    public void testSelectStaticColumnWithMetadata()
    {
        Consumer<ColumnFilter> check = filter -> {
            testRoundTrips(filter);
            assertFetchedQueried(true, true, filter, s1);

            assertEquals("*/*", filter.toString());
            assertEquals("s1", filter.toCQLString());
            assertFetchedQueried(true, true, filter, v1, v2, s2);
            assertCellFetchedQueried(true, true, filter, v2, path0, path1, path2, path3, path4);
            assertCellFetchedQueried(true, true, filter, s2, path0, path1, path2, path3, path4);
        };

        check.accept(ColumnFilter.selection(metadata, PartitionColumns.builder().add(s1).build()));
        check.accept(ColumnFilter.allColumnsBuilder(metadata).add(s1).build());
    }

    @Test
    public void testSelectCellWithMetadata()
    {
        ColumnFilter filter = ColumnFilter.allColumnsBuilder(metadata).select(v2, path1).build();
        testRoundTrips(filter);
        assertFetchedQueried(true, true, filter, v2);

        assertEquals("*/*", filter.toString());
        assertEquals("v2[1]", filter.toCQLString());
        assertFetchedQueried(true, true, filter, s1, s2, v1);
        assertCellFetchedQueried(true, true, filter, v2, path1);
        assertCellFetchedQueried(true, false, filter, v2, path0, path2, path3, path4);
        assertCellFetchedQueried(true, true, filter, s2, path0, path1, path2, path3, path4);
    }

    @Test
    public void testSelectStaticColumnCellWithMetadata()
    {
        ColumnFilter filter = ColumnFilter.allColumnsBuilder(metadata).select(s2, path1).build();
        testRoundTrips(filter);
        assertFetchedQueried(true, true, filter, s2);

        assertEquals("*/*", filter.toString());
        assertEquals("s2[1]", filter.toCQLString());
        assertFetchedQueried(true, true, filter, v1, v2, s1);
        assertCellFetchedQueried(true, true, filter, v2, path0, path1, path2, path3, path4);
        assertCellFetchedQueried(true, true, filter, s2, path1);
        assertCellFetchedQueried(true, false, filter, s2, path0, path2, path3, path4);
    }

    private void testRoundTrips(ColumnFilter cf)
    {
        testRoundTrip(cf, MessagingService.VERSION_30);
        testRoundTrip(cf, MessagingService.VERSION_3014);
    }

    private void testRoundTrip(ColumnFilter columnFilter, int version)
    {
        try
        {
            DataOutputBuffer output = new DataOutputBuffer();
            serializer.serialize(columnFilter, output, version);
            Assert.assertEquals(serializer.serializedSize(columnFilter, version), output.position());
            DataInputPlus input = new DataInputBuffer(output.buffer(), false);
            ColumnFilter deserialized = serializer.deserialize(input, version, metadata);
            Assert.assertEquals(deserialized, columnFilter);
        }
        catch (IOException e)
        {
            throw Throwables.propagate(e);
        }
    }

    private static void assertFetchedQueried(boolean expectedIncluded,
                                             boolean expectedNotSkipped,
                                             ColumnFilter filter,
                                             ColumnDefinition... columns)
    {
        for (ColumnDefinition column : columns)
        {
            assertEquals(String.format("Expected includes(%s) to be %s", column.name, expectedIncluded),
                         expectedIncluded, filter.includes(column));
            if (expectedIncluded)
                assertEquals(String.format("Expected canSkipValue(%s) to be %s", column.name, !expectedNotSkipped),
                             !expectedNotSkipped, filter.canSkipValue(column));
        }
    }

    private static void assertCellFetchedQueried(boolean expectedIncluded,
                                                 boolean expectedNotSkipped,
                                                 ColumnFilter filter,
                                                 ColumnDefinition column,
                                                 CellPath... paths)
    {
        ColumnFilter.Tester tester = filter.newTester(column);

        for (CellPath path : paths)
        {
            int p = ByteBufferUtil.toInt(path.get(0));

            if (tester != null)
            {
                assertEquals(String.format("Expected tester.includes(%s:%s) to be %s", column.name, p, expectedIncluded),
                             expectedIncluded, tester.includes(path));
                if (expectedIncluded)
                    assertEquals(String.format("Expected tester.canSkipValue(%s:%s) to be %s", column.name, p, !expectedNotSkipped),
                                 !expectedNotSkipped, tester.canSkipValue(path));
            }
        }
    }
}