content: /*
   * 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.cql3.validation.operations;
  
  import java.io.BufferedOutputStream;
  import java.io.IOException;
  import java.io.OutputStreamWriter;
  import java.io.Writer;
  import java.nio.ByteBuffer;
  import java.nio.charset.StandardCharsets;
  import java.nio.file.Files;
  import java.util.UUID;
  
  import com.google.common.base.StandardSystemProperty;
  import org.junit.Test;
  
  import com.datastax.driver.core.exceptions.InvalidQueryException;
  import org.apache.cassandra.Util;
  import org.apache.cassandra.cql3.CQLTester;
  import org.apache.cassandra.io.util.File;
  import org.apache.cassandra.tools.ToolRunner;
  import org.apache.cassandra.utils.FBUtilities;
  import org.assertj.core.api.Assertions;
  
  import static org.apache.cassandra.utils.ByteBufferUtil.EMPTY_BYTE_BUFFER;
  
  public class InsertInvalidateSizedRecordsTest extends CQLTester
  {
      private static final ByteBuffer LARGE_BLOB = ByteBuffer.allocate(FBUtilities.MAX_UNSIGNED_SHORT + 1);
      private static final ByteBuffer MEDIUM_BLOB = ByteBuffer.allocate(FBUtilities.MAX_UNSIGNED_SHORT / 2 + 10);
  
      {
          requireNetwork();
      }
  
      @Test
      public void singleValuePk()
      {
          createTable(KEYSPACE, "CREATE TABLE %s (a blob PRIMARY KEY)");
          Assertions.assertThatThrownBy(() -> executeNet("INSERT INTO %s (a) VALUES (?)", LARGE_BLOB))
                    .hasRootCauseInstanceOf(InvalidQueryException.class)
                    .hasRootCauseMessage("Key length of " + LARGE_BLOB.remaining() + " is longer than maximum of 65535");
  
          // null / empty checks
          Assertions.assertThatThrownBy(() -> executeNet("INSERT INTO %s (a) VALUES (?)", new Object[] {null}))
                    .hasRootCauseInstanceOf(InvalidQueryException.class)
                    .hasRootCauseMessage("Invalid null value for column a");
          Assertions.assertThatThrownBy(() -> executeNet("INSERT INTO %s (a) VALUES (?)", EMPTY_BYTE_BUFFER))
                    .hasRootCauseInstanceOf(InvalidQueryException.class)
                    .hasRootCauseMessage("Key may not be empty");
      }
  
      @Test
      public void compositeValuePk()
      {
          createTable(KEYSPACE, "CREATE TABLE %s (a blob, b blob, PRIMARY KEY ((a, b)))");
          // sum of columns is too large
          Assertions.assertThatThrownBy(() -> executeNet("INSERT INTO %s (a, b) VALUES (?, ?)", MEDIUM_BLOB, MEDIUM_BLOB))
                    .hasRootCauseInstanceOf(InvalidQueryException.class)
                    .hasRootCauseMessage("Key length of " + (MEDIUM_BLOB.remaining() * 2) + " is longer than maximum of 65535");
  
          // single column is too large
          Assertions.assertThatThrownBy(() -> executeNet("INSERT INTO %s (a, b) VALUES (?, ?)", MEDIUM_BLOB, LARGE_BLOB))
                    .hasRootCauseInstanceOf(InvalidQueryException.class)
                    .hasRootCauseMessage("Key length of " + (MEDIUM_BLOB.remaining() + LARGE_BLOB.remaining()) + " is longer than maximum of 65535");
  
          // null / empty checks
          // this is an inconsistent behavior... null is blocked by org.apache.cassandra.db.MultiCBuilder.OneClusteringBuilder.addElementToAll
          // but this does not count empty as null, and doesn't check for this case...  We have a requirement in cqlsh that empty is allowed when
          // user opts-in to allow it (NULL='-'), so we will find that null is blocked, but empty is allowed!
          Assertions.assertThatThrownBy(() -> executeNet("INSERT INTO %s (a, b) VALUES (?, ?)", new Object[] {null, null}))
                    .hasRootCauseInstanceOf(InvalidQueryException.class)
                    .hasRootCauseMessage("Invalid null value for column a");
          Assertions.assertThatThrownBy(() -> executeNet("INSERT INTO %s (a, b) VALUES (?, ?)", new Object[] {MEDIUM_BLOB, null}))
                    .hasRootCauseInstanceOf(InvalidQueryException.class)
                    .hasRootCauseMessage("Invalid null value for column b");
          Assertions.assertThatThrownBy(() -> executeNet("INSERT INTO %s (a, b) VALUES (?, ?)", new Object[] {null, MEDIUM_BLOB}))
                    .hasRootCauseInstanceOf(InvalidQueryException.class)
                    .hasRootCauseMessage("Invalid null value for column a");
  
          // empty is allowed when composite partition columns...
          executeNet("INSERT INTO %s (a, b) VALUES (?, ?)", EMPTY_BYTE_BUFFER, EMPTY_BYTE_BUFFER);
          execute("TRUNCATE %s");
  
          executeNet("INSERT INTO %s (a, b) VALUES (?, ?)", MEDIUM_BLOB, EMPTY_BYTE_BUFFER);
          execute("TRUNCATE %s");
  
          executeNet("INSERT INTO %s (a, b) VALUES (?, ?)", EMPTY_BYTE_BUFFER, MEDIUM_BLOB);
      }
  
      @Test
      public void singleValueClustering()
      {
          createTable(KEYSPACE, "CREATE TABLE %s (a blob, b blob, PRIMARY KEY (a, b))");
          Assertions.assertThatThrownBy(() -> executeNet("INSERT INTO %s (a, b) VALUES (?, ?)", MEDIUM_BLOB, LARGE_BLOB))
                    .hasRootCauseInstanceOf(InvalidQueryException.class)
                    .hasRootCauseMessage("Key length of " + LARGE_BLOB.remaining() + " is longer than maximum of 65535");
  
          // null / empty checks
          Assertions.assertThatThrownBy(() -> executeNet("INSERT INTO %s (a, b) VALUES (?, ?)", MEDIUM_BLOB, null))
                    .hasRootCauseInstanceOf(InvalidQueryException.class)
                    .hasRootCauseMessage("Invalid null value for column b");
  
          // org.apache.cassandra.db.MultiCBuilder.OneClusteringBuilder.addElementToAll defines "null" differently than most of the code;
          // most of the code defines null as:
          //   value == null || accessor.isEmpty(value)
          // but the code defines null as
          //   value == null
          // In CASSANDRA-18504 a new isNull method was added to the type, as blob and text both "should" allow empty, but this scattered null logic doesn't allow...
          // For backwards compatability reasons, need to keep empty support
          executeNet("INSERT INTO %s (a, b) VALUES (?, ?)", MEDIUM_BLOB, EMPTY_BYTE_BUFFER);
      }
  
      @Test
      public void compositeValueClustering()
      {
          createTable(KEYSPACE, "CREATE TABLE %s (a blob, b blob, c blob, PRIMARY KEY (a, b, c))");
          // sum of columns is too large
          Assertions.assertThatThrownBy(() -> executeNet("INSERT INTO %s (a, b, c) VALUES (?, ?, ?)", MEDIUM_BLOB, MEDIUM_BLOB, MEDIUM_BLOB))
                    .hasRootCauseInstanceOf(InvalidQueryException.class)
                    .hasRootCauseMessage("Key length of " + (MEDIUM_BLOB.remaining() * 2) + " is longer than maximum of 65535");
  
          // single column is too large
          // the logic prints the total clustering size and not the single column's size that was too large
          Assertions.assertThatThrownBy(() -> executeNet("INSERT INTO %s (a, b, c) VALUES (?, ?, ?)", MEDIUM_BLOB, MEDIUM_BLOB, LARGE_BLOB))
                    .hasRootCauseInstanceOf(InvalidQueryException.class)
                    .hasRootCauseMessage("Key length of " + (MEDIUM_BLOB.remaining() + LARGE_BLOB.remaining()) + " is longer than maximum of 65535");
      }
  
      @Test
      public void singleValueIndex()
      {
          Util.assumeLegacySecondaryIndex();
          createTable(KEYSPACE, "CREATE TABLE %s (a blob, b blob, PRIMARY KEY (a))");
          String table = KEYSPACE + "." + currentTable();
          execute("CREATE INDEX single_value_index ON %s (b)");
          Assertions.assertThatThrownBy(() -> executeNet("INSERT INTO %s (a, b) VALUES (?, ?)", MEDIUM_BLOB, LARGE_BLOB))
                    .hasRootCauseInstanceOf(InvalidQueryException.class)
                    .hasRootCauseMessage("Cannot index value of size " + LARGE_BLOB.remaining() + " for index single_value_index on " + table + "(b) (maximum allowed size=65535)");
      }
  
      /**
       * Partial port of cqlsh_tests.test_cqlsh_copy.TestCqlshCopy#test_reading_empty_strings_for_different_types
       */
      @Test
      public void readingEmptyStringsForDifferentTypes() throws IOException
      {
          createTable(KEYSPACE, "CREATE TABLE %s (\n" +
                                "                a text,\n" +
                                "                b text,\n" +
                                "                c text,\n" +
                                "                d text,\n" +
                                "                o uuid,\n" +
                                "                 i1 bigint,\n" +
                                "                 i2 bigint,\n" +
                                "                 t text,\n" +
                                "                 i3 bigint,\n" +
                                "                 PRIMARY KEY ((a, b, c, d), o)\n" +
                                "            )");
  
          File csv = new File(Files.createTempFile(new File(StandardSystemProperty.JAVA_IO_TMPDIR.value()).toPath(), "test_reading_empty_strings_for_different_types", ".csv"));
          try (Writer out = new OutputStreamWriter(new BufferedOutputStream(csv.newOutputStream(File.WriteMode.OVERWRITE)), StandardCharsets.UTF_8))
          {
              out.write(",,,a1,645e7d3c-aef7-4e3c-b834-24b792cf2e55,,,,r1\n");
          }
          String table = KEYSPACE + "." + currentTable();
          String template = "COPY %s FROM '%s' WITH NULL='-' AND PREPAREDSTATEMENTS = %s";
          // This is different from cqlsh_tests.test_cqlsh_copy.TestCqlshCopy#test_reading_empty_strings_for_different_types as "false" actually is broken!
          // If you do false, then the parsing actually fails... and the test didn't actually check the return code from cqlsh so doesn't detect that it failed!
          // What is worse is the test didn't truncate either, which means that the "false" case is reading the "true" case data!
          // Rather than try to fix the test and cqlsh... this test only keeps the "true" logic as it is what is needed to make sure CASSANDRA-18504
          // didn't break things.
          ToolRunner.invokeCqlsh(String.format(template, table, csv.absolutePath(), true)).assertOnCleanExit();
          assertRowsNet(executeNet("SELECT * FROM %s"), row("", "", "", "a1", UUID.fromString("645e7d3c-aef7-4e3c-b834-24b792cf2e55"), null, null, null, "r1"));
      }
  }
  