text: // 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.
  
  #include "kudu/common/row_operations.h"
  
  #include <cstdint>
  #include <cstdlib>
  #include <memory>
  #include <ostream>
  #include <string>
  #include <vector>
  
  #include <gflags/gflags_declare.h>
  #include <glog/logging.h>
  #include <gtest/gtest.h>
  
  #include "kudu/common/common.pb.h"
  #include "kudu/common/partial_row.h"
  #include "kudu/common/row.h"
  #include "kudu/common/schema.h"
  #include "kudu/common/types.h"
  #include "kudu/common/wire_protocol.pb.h"
  #include "kudu/gutil/basictypes.h"
  #include "kudu/gutil/dynamic_annotations.h"
  #include "kudu/gutil/macros.h"
  #include "kudu/gutil/strings/substitute.h"
  #include "kudu/util/bitmap.h"
  #include "kudu/util/memory/arena.h"
  #include "kudu/util/slice.h"
  #include "kudu/util/status.h"
  #include "kudu/util/test_macros.h"
  #include "kudu/util/test_util.h"
  
  using std::shared_ptr;
  using std::string;
  using std::vector;
  using strings::Substitute;
  
  DECLARE_int32(max_cell_size_bytes);
  
  namespace kudu {
  
  class RowOperationsTest : public KuduTest {
   public:
    RowOperationsTest()
      : arena_(1024) {
      SeedRandom();
  
      SchemaBuilder builder;
      CHECK_OK(builder.AddKeyColumn("key", INT32));
      CHECK_OK(builder.AddColumn("int_val", INT32));
      CHECK_OK(builder.AddNullableColumn("string_val", STRING));
      schema_ = builder.Build();
      schema_without_ids_ = builder.BuildWithoutIds();
    }
   protected:
    void CheckDecodeDoesntCrash(const Schema& client_schema,
                                const Schema& server_schema,
                                const RowOperationsPB& pb);
    void DoFuzzTest(const Schema& server_schema,
                    const KuduPartialRow& row,
                    int n_random_changes);
  
    Schema schema_;
    Schema schema_without_ids_;
    Arena arena_;
  };
  
  // Perform some random mutation to a random byte in the provided string.
  static void DoRandomMutation(string* s) {
    int target_idx = random() % s->size();
    char* target_byte = &(*s)[target_idx];
    switch (random() % 3) {
      case 0:
        // increment a random byte by 1
        (*target_byte)++;
        break;
      case 1:
        // decrement a random byte by 1
        (*target_byte)--;
        break;
      case 2:
        // replace byte with random value
        (*target_byte) = random();
        break;
    }
  }
  
  void RowOperationsTest::CheckDecodeDoesntCrash(const Schema& client_schema,
                                                 const Schema& server_schema,
                                                 const RowOperationsPB& pb) {
    arena_.Reset();
    RowOperationsPBDecoder decoder(&pb, &client_schema, &server_schema, &arena_);
    // Decoding the operations, regardless of the mode, should not result in a
    // crash.
    vector<DecodedRowOperation> ops;
    Status s = decoder.DecodeOperations<DecoderMode::WRITE_OPS>(&ops);
    if (!s.ok()) {
      s = decoder.DecodeOperations<DecoderMode::SPLIT_ROWS>(&ops);
    }
    if (s.ok() && !ops.empty()) {
      // If we got an OK result, then we should be able to stringify without
      // crashing. This ensures that any indirect data (eg strings) gets
      // set correctly.
      ignore_result(ops[0].ToString(server_schema));
    }
    // Bad Status is OK -- we expect corruptions here.
  }
  
  void RowOperationsTest::DoFuzzTest(const Schema& server_schema,
                                     const KuduPartialRow& row,
                                     int n_random_changes) {
    for (int operation = 0; operation <= 8; operation++) {
      RowOperationsPB pb;
      RowOperationsPBEncoder enc(&pb);
  
      switch (operation) {
        case 0:
          enc.Add(RowOperationsPB::INSERT, row);
          break;
        case 1:
          enc.Add(RowOperationsPB::UPSERT, row);
          break;
        case 2:
          enc.Add(RowOperationsPB::UPDATE, row);
          break;
        case 3:
          enc.Add(RowOperationsPB::DELETE, row);
          break;
        case 4:
          enc.Add(RowOperationsPB::SPLIT_ROW, row);
          break;
        case 5:
          enc.Add(RowOperationsPB::RANGE_LOWER_BOUND, row);
          break;
        case 6:
          enc.Add(RowOperationsPB::RANGE_UPPER_BOUND, row);
          break;
        case 7:
          enc.Add(RowOperationsPB::EXCLUSIVE_RANGE_LOWER_BOUND, row);
          break;
        case 8:
          enc.Add(RowOperationsPB::INCLUSIVE_RANGE_UPPER_BOUND, row);
          break;
      }
  
      const Schema* client_schema = row.schema();
  
      // Check that the un-mutated row doesn't crash.
      CheckDecodeDoesntCrash(*client_schema, server_schema, pb);
  
      RowOperationsPB mutated;
  
      // Check all possible truncations of the protobuf 'rows' field.
      for (int i = 0; i < pb.rows().size(); i++) {
        mutated.CopyFrom(pb);
        mutated.mutable_rows()->resize(i);
        CheckDecodeDoesntCrash(*client_schema, server_schema, mutated);
      }
  
      // Check bit flips of every bit in the first three bytes, which are
      // particularly interesting, since they contain the null/isset
      // bitmaps.
      for (int bit = 0; bit < 8 * 3; bit++) {
        int byte_idx = bit / 8;
        int bit_idx = bit % 8;
        int mask = 1 << bit_idx;
  
        (*mutated.mutable_rows())[byte_idx] ^= mask;
        CheckDecodeDoesntCrash(*client_schema, server_schema, mutated);
        (*mutated.mutable_rows())[byte_idx] ^= mask;
      }
  
      // Check random byte changes in the 'rows' field.
      for (int i = 0; i < n_random_changes; i++) {
        mutated.CopyFrom(pb);
        DoRandomMutation(mutated.mutable_rows());
        CheckDecodeDoesntCrash(*client_schema, server_schema, mutated);
      }
    }
  }
  
  // Test that, even if the protobuf is corrupt in some way, we do not
  // crash. These protobufs are provided by clients, so we want to make sure
  // a malicious client can't crash the server.
  TEST_F(RowOperationsTest, FuzzTest) {
    const int n_iters = AllowSlowTests() ? 10000 : 1000;
  
    KuduPartialRow row(&schema_without_ids_);
    EXPECT_OK(row.SetInt32("int_val", 54321));
    EXPECT_OK(row.SetStringCopy("string_val", "hello world"));
    DoFuzzTest(schema_, row, n_iters);
    EXPECT_OK(row.SetNull("string_val"));
    DoFuzzTest(schema_, row, n_iters);
  }
  
  // Add the given column, but with some probability change the type
  // and nullability.
  void AddFuzzedColumn(SchemaBuilder* builder,
                       const string& name,
                       DataType default_type) {
    DataType rand_types[] = {INT32, INT64, DOUBLE, STRING};
    DataType t = default_type;
    if (random() % 3 == 0) {
      t = rand_types[random() % arraysize(rand_types)];
    }
    bool nullable = random() & 1;
    CHECK_OK(builder->AddColumn(name, t, nullable, NULL, NULL));
  }
  
  // Generate a randomized schema, where some columns might be missing,
  // and types/nullability are randomized. We weight towards not making
  // too many changes so that it's likely we generate compatible client
  // and server schemas.
  Schema GenRandomSchema(bool with_ids) {
    SchemaBuilder builder;
    if (random() % 5 != 0) {
      AddFuzzedColumn(&builder, "c1", INT32);
    }
    if (random() % 5 != 0) {
      AddFuzzedColumn(&builder, "c2", INT32);
    }
    if (random() % 5 != 0 || !builder.is_valid()) {
      AddFuzzedColumn(&builder, "c3", STRING);
    }
  
    return with_ids ? builder.Build() : builder.BuildWithoutIds();
  }
  
  namespace {
  
  struct FailingCase {
    Schema* client_schema;
    Schema* server_schema;
    KuduPartialRow* row;
  };
  FailingCase g_failing_case;
  
  // ASAN callback which will dump the case which caused a failure.
  void DumpFailingCase() {
    LOG(INFO) << "Failed on the following case:";
    LOG(INFO) << "Client schema:\n" << g_failing_case.client_schema->ToString();
    LOG(INFO) << "Server schema:\n" << g_failing_case.server_schema->ToString();
    LOG(INFO) << "Row: " << g_failing_case.row->ToString();
  }
  
  void GlogFailure() {
    DumpFailingCase();
    abort();
  }
  
  } // anonymous namespace
  
  // Fuzz test which generates random pairs of client/server schemas, with
  // random mutations like adding an extra column, removing a column, changing
  // types, and changing nullability.
  TEST_F(RowOperationsTest, SchemaFuzz) {
  #ifdef ADDRESS_SANITIZER
    // Prevent timeouts in ASAN build.
    const int kSlowTestIters = 1000;
  #else
    const int kSlowTestIters = 10000;
  #endif
    const int n_iters = AllowSlowTests() ? kSlowTestIters : 10;
    for (int i = 0; i < n_iters; i++) {
      // Generate a random client and server schema pair.
      Schema client_schema = GenRandomSchema(false);
      Schema server_schema = GenRandomSchema(true);
      KuduPartialRow row(&client_schema);
  
      // On a crash or ASAN failure, dump the case information to the log so we
      // can write a more specific repro.
      g_failing_case.client_schema = &client_schema;
      g_failing_case.server_schema = &server_schema;
      g_failing_case.row = &row;
      ASAN_SET_DEATH_CALLBACK(&DumpFailingCase);
      google::InstallFailureFunction(&GlogFailure);
  
      for (int i = 0; i < client_schema.num_columns(); i++) {
        if (client_schema.column(i).is_nullable() &&
            random() % 3 == 0) {
          CHECK_OK(row.SetNull(i));
          continue;
        }
        switch (client_schema.column(i).type_info()->type()) {
          case INT32:
            CHECK_OK(row.SetInt32(i, 12345));
            break;
          case INT64:
            CHECK_OK(row.SetInt64(i, 12345678));
            break;
          case DOUBLE:
            CHECK_OK(row.SetDouble(i, 1234.5678));
            break;
          case STRING:
            CHECK_OK(row.SetStringCopy(i, "hello"));
            break;
          default:
            LOG(FATAL);
        }
      }
  
      DoFuzzTest(server_schema, row, 100);
      ASAN_SET_DEATH_CALLBACK(NULL);
      google::InstallFailureFunction(&abort);
    }
  }
  
  // One case from SchemaFuzz which failed previously.
  TEST_F(RowOperationsTest, TestFuzz1) {
    SchemaBuilder client_schema_builder;
    client_schema_builder.AddColumn("c1", INT32, false, nullptr, nullptr);
    client_schema_builder.AddColumn("c2", STRING, false, nullptr, nullptr);
    Schema client_schema = client_schema_builder.BuildWithoutIds();
    SchemaBuilder server_schema_builder;
    server_schema_builder.AddColumn("c1", INT32, false, nullptr, nullptr);
    server_schema_builder.AddColumn("c2", STRING, false, nullptr, nullptr);
    Schema server_schema = server_schema_builder.Build();
    KuduPartialRow row(&client_schema);
    CHECK_OK(row.SetInt32(0, 12345));
    CHECK_OK(row.SetStringCopy(1, "hello"));
    DoFuzzTest(server_schema, row, 100);
  }
  
  // Another case from SchemaFuzz which failed previously.
  TEST_F(RowOperationsTest, TestFuzz2) {
    SchemaBuilder client_schema_builder;
    client_schema_builder.AddColumn("c1", STRING, true, nullptr, nullptr);
    client_schema_builder.AddColumn("c2", STRING, false, nullptr, nullptr);
    Schema client_schema = client_schema_builder.BuildWithoutIds();
    SchemaBuilder server_schema_builder;
    server_schema_builder.AddColumn("c1", STRING, true, nullptr, nullptr);
    server_schema_builder.AddColumn("c2", STRING, false, nullptr, nullptr);
    Schema server_schema = server_schema_builder.Build();
    KuduPartialRow row(&client_schema);
    CHECK_OK(row.SetNull(0));
    CHECK_OK(row.SetStringCopy(1, "hello"));
    DoFuzzTest(server_schema, row, 100);
  }
  
  namespace {
  
  // Project client_row into server_schema, and stringify the result.
  // If an error occurs, the result string is "error: <stringified Status>"
  string TestProjection(RowOperationsPB::Type type,
                        const KuduPartialRow& client_row,
                        const Schema& server_schema) {
    RowOperationsPB pb;
    RowOperationsPBEncoder enc(&pb);
    enc.Add(type, client_row);
  
    // Decode it
    Arena arena(1024);
    vector<DecodedRowOperation> ops;
    RowOperationsPBDecoder dec(&pb, client_row.schema(), &server_schema, &arena);
    Status s = dec.DecodeOperations<DecoderMode::WRITE_OPS>(&ops);
  
    if (!s.ok()) {
      return "error: " + s.ToString();
    }
    CHECK_EQ(1, ops.size());
    return ops[0].ToString(server_schema);
  }
  
  } // anonymous namespace
  
  // Test decoding partial rows from a client who has a schema which matches
  // the table schema.
  TEST_F(RowOperationsTest, ProjectionTestWholeSchemaSpecified) {
    Schema client_schema({ ColumnSchema("key", INT32),
                           ColumnSchema("int_val", INT32),
                           ColumnSchema("string_val", STRING, true) },
                         1);
    // Test a row missing 'key', which is key column.
    {
      KuduPartialRow client_row(&client_schema);
      EXPECT_EQ("row error: Invalid argument: No value provided for required column: "
                "key INT32 NOT NULL",
                TestProjection(RowOperationsPB::INSERT, client_row, schema_));
    }
  
    // Force to set null on key column.
    {
      int col_idx;
      ASSERT_OK(client_schema.FindColumn("key", &col_idx));
      KuduPartialRow client_row(&client_schema);
      ContiguousRow row(&client_schema, client_row.row_data_);
      row.set_null(col_idx, true);
      BitmapSet(client_row.isset_bitmap_, col_idx);
      EXPECT_EQ("row error: Invalid argument: NULL values not allowed for non-nullable column: "
                "key INT32 NOT NULL",
                TestProjection(RowOperationsPB::INSERT, client_row, schema_));
    }
  
    // Test a row missing 'int_val', which is required.
    {
      KuduPartialRow client_row(&client_schema);
      CHECK_OK(client_row.SetInt32("key", 12345));
      EXPECT_EQ("row error: Invalid argument: No value provided for required column: "
                "int_val INT32 NOT NULL",
                TestProjection(RowOperationsPB::INSERT, client_row, schema_));
    }
  
    // Test a row missing 'string_val', which is nullable
    {
      KuduPartialRow client_row(&client_schema);
      CHECK_OK(client_row.SetInt32("key", 12345));
      CHECK_OK(client_row.SetInt32("int_val", 54321));
      // The NULL should get filled in.
      EXPECT_EQ("INSERT (int32 key=12345, int32 int_val=54321, string string_val=NULL)",
                TestProjection(RowOperationsPB::INSERT, client_row, schema_));
    }
  
    // Test a row with all of the fields specified, both with the nullable field
    // specified to be NULL and non-NULL.
    {
      KuduPartialRow client_row(&client_schema);
      CHECK_OK(client_row.SetInt32("key", 12345));
      CHECK_OK(client_row.SetInt32("int_val", 54321));
      CHECK_OK(client_row.SetStringCopy("string_val", "hello world"));
      EXPECT_EQ(R"(INSERT (int32 key=12345, int32 int_val=54321, string string_val="hello world"))",
                TestProjection(RowOperationsPB::INSERT, client_row, schema_));
  
      // The first result should have the field specified.
      // The second result should have the field NULL, since it was explicitly set.
      CHECK_OK(client_row.SetNull("string_val"));
      EXPECT_EQ("INSERT (int32 key=12345, int32 int_val=54321, string string_val=NULL)",
                TestProjection(RowOperationsPB::INSERT, client_row, schema_));
  
    }
  }
  
  TEST_F(RowOperationsTest, ProjectionTestWithDefaults) {
    int32_t nullable_default = 123;
    int32_t non_null_default = 456;
    SchemaBuilder b;
    CHECK_OK(b.AddKeyColumn("key", INT32));
    CHECK_OK(b.AddColumn("nullable_with_default", INT32, true,
                         &nullable_default, &nullable_default));
    CHECK_OK(b.AddColumn("non_null_with_default", INT32, false,
                         &non_null_default, &non_null_default));
    Schema server_schema = b.Build();
  
    // Clients may not have the defaults specified.
    // TODO: evaluate whether this should be true - how "dumb" should clients be?
    Schema client_schema({ ColumnSchema("key", INT32),
                           ColumnSchema("nullable_with_default", INT32, true),
                           ColumnSchema("non_null_with_default", INT32, false) },
                         1);
  
    // Specify just the key. The other two columns have defaults, so they'll get filled in.
    {
      KuduPartialRow client_row(&client_schema);
      CHECK_OK(client_row.SetInt32("key", 12345));
      EXPECT_EQ("INSERT (int32 key=12345, int32 nullable_with_default=123,"
                " int32 non_null_with_default=456)",
                TestProjection(RowOperationsPB::INSERT, client_row, server_schema));
    }
  
    // Specify the key and override both defaults
    {
      KuduPartialRow client_row(&client_schema);
      CHECK_OK(client_row.SetInt32("key", 12345));
      CHECK_OK(client_row.SetInt32("nullable_with_default", 12345));
      CHECK_OK(client_row.SetInt32("non_null_with_default", 54321));
      EXPECT_EQ("INSERT (int32 key=12345, int32 nullable_with_default=12345,"
                " int32 non_null_with_default=54321)",
                TestProjection(RowOperationsPB::INSERT, client_row, server_schema));
    }
  
    // Specify the key and override both defaults, overriding the nullable
    // one to NULL.
    {
      KuduPartialRow client_row(&client_schema);
      CHECK_OK(client_row.SetInt32("key", 12345));
      CHECK_OK(client_row.SetNull("nullable_with_default"));
      CHECK_OK(client_row.SetInt32("non_null_with_default", 54321));
      EXPECT_EQ("INSERT (int32 key=12345, int32 nullable_with_default=NULL,"
                " int32 non_null_with_default=54321)",
                TestProjection(RowOperationsPB::INSERT, client_row, server_schema));
    }
  
    // Specify the key and override both defaults, overriding the non-nullable
    // one to NULL.
    {
      KuduPartialRow client_row(&client_schema);
      CHECK_OK(client_row.SetInt32("key", 12345));
      CHECK_OK(client_row.SetInt32("nullable_with_default", 12345));
      Status s = client_row.SetNull("non_null_with_default");
      CHECK(s.IsInvalidArgument());
      EXPECT_EQ("INSERT (int32 key=12345, int32 nullable_with_default=12345,"
                " int32 non_null_with_default=456)",
                TestProjection(RowOperationsPB::INSERT, client_row, server_schema));
    }
  }
  
  // Test cases where the client only has a subset of the fields
  // of the table, but where the missing columns have defaults
  // or are NULLable.
  TEST_F(RowOperationsTest, ProjectionTestWithClientHavingValidSubset) {
    int32_t nullable_default = 123;
    SchemaBuilder b;
    CHECK_OK(b.AddKeyColumn("key", INT32));
    CHECK_OK(b.AddColumn("int_val", INT32));
    CHECK_OK(b.AddColumn("new_int_with_default", INT32, false,
                         &nullable_default, &nullable_default));
    CHECK_OK(b.AddNullableColumn("new_nullable_int", INT32));
    Schema server_schema = b.Build();
  
    Schema client_schema({ ColumnSchema("key", INT32),
                           ColumnSchema("int_val", INT32) },
                         1);
  
    // Specify just the key. This is an error because we're missing int_val.
    {
      KuduPartialRow client_row(&client_schema);
      CHECK_OK(client_row.SetInt32("key", 12345));
      EXPECT_EQ("row error: Invalid argument: No value provided for required column:"
                " int_val INT32 NOT NULL",
                TestProjection(RowOperationsPB::INSERT, client_row, server_schema));
    }
  
    // Specify both of the columns that the client is aware of.
    // Defaults should be filled for the other two.
    {
      KuduPartialRow client_row(&client_schema);
      CHECK_OK(client_row.SetInt32("key", 12345));
      CHECK_OK(client_row.SetInt32("int_val", 12345));
      EXPECT_EQ("INSERT (int32 key=12345, int32 int_val=12345,"
                " int32 new_int_with_default=123, int32 new_nullable_int=NULL)",
                TestProjection(RowOperationsPB::INSERT, client_row, server_schema));
    }
  }
  
  // Test cases where the client is missing a column which is non-null
  // and has no default. This is an incompatible client.
  TEST_F(RowOperationsTest, ProjectionTestWithClientHavingInvalidSubset) {
    SchemaBuilder b;
    CHECK_OK(b.AddKeyColumn("key", INT32));
    CHECK_OK(b.AddColumn("int_val", INT32));
    Schema server_schema = b.Build();
  
    CHECK_OK(b.RemoveColumn("int_val"));
    Schema client_schema = b.BuildWithoutIds();
  
    {
      KuduPartialRow client_row(&client_schema);
      CHECK_OK(client_row.SetInt32("key", 12345));
      EXPECT_EQ("error: Invalid argument: Client missing required column:"
                " int_val INT32 NOT NULL",
                TestProjection(RowOperationsPB::INSERT, client_row, server_schema));
    }
  }
  
  // Simple Update case where the client and server schemas match.
  TEST_F(RowOperationsTest, TestProjectUpdates) {
    Schema client_schema({ ColumnSchema("key", INT32),
                           ColumnSchema("int_val", INT32),
                           ColumnSchema("string_val", STRING, true) },
                         1);
    Schema server_schema = SchemaBuilder(client_schema).Build();
  
    // Check without specifying any columns
    KuduPartialRow client_row(&client_schema);
    EXPECT_EQ("row error: Invalid argument: No value provided for key column: key INT32 NOT NULL",
              TestProjection(RowOperationsPB::UPDATE, client_row, server_schema));
  
    // Specify the key and no columns to update
    ASSERT_OK(client_row.SetInt32("key", 12345));
    EXPECT_EQ("row error: Invalid argument: No fields updated, key is: (int32 key=12345)",
              TestProjection(RowOperationsPB::UPDATE, client_row, server_schema));
  
    // Specify the key and update one column.
    ASSERT_OK(client_row.SetInt32("int_val", 12345));
    EXPECT_EQ("MUTATE (int32 key=12345) SET int_val=12345",
              TestProjection(RowOperationsPB::UPDATE, client_row, server_schema));
  
    // Specify the key and update both columns
    ASSERT_OK(client_row.SetStringNoCopy("string_val", "foo"));
    EXPECT_EQ(R"(MUTATE (int32 key=12345) SET int_val=12345, string_val="foo")",
              TestProjection(RowOperationsPB::UPDATE, client_row, server_schema));
  
    // Update the nullable column to null.
    ASSERT_OK(client_row.SetNull("string_val"));
    EXPECT_EQ("MUTATE (int32 key=12345) SET int_val=12345, string_val=NULL",
              TestProjection(RowOperationsPB::UPDATE, client_row, server_schema));
  
    // Force to set null on key column.
    {
      KuduPartialRow client_row(&client_schema);
      int col_idx;
      ASSERT_OK(client_schema.FindColumn("key", &col_idx));
      ContiguousRow row(&client_schema, client_row.row_data_);
      row.set_null(col_idx, true);
      BitmapSet(client_row.isset_bitmap_, col_idx);
      EXPECT_EQ("row error: Invalid argument: NULL values not allowed for key column: "
                "key INT32 NOT NULL",
                TestProjection(RowOperationsPB::UPDATE, client_row, server_schema));
    }
  
    // Force to set null on non-nullable column.
    {
      KuduPartialRow client_row(&client_schema);
      ASSERT_OK(client_row.SetInt32("key", 12345));
      int col_idx;
      ASSERT_OK(client_schema.FindColumn("int_val", &col_idx));
      ContiguousRow row(&client_schema, client_row.row_data_);
      row.set_null(col_idx, true);
      BitmapSet(client_row.isset_bitmap_, col_idx);
      EXPECT_EQ("row error: Invalid argument: NULL value not allowed for non-nullable column: "
                "int_val INT32 NOT NULL",
                TestProjection(RowOperationsPB::UPDATE, client_row, server_schema));
    }
  }
  
  // Client schema has the columns in a different order. Makes
  // sure the name-based projection is functioning.
  TEST_F(RowOperationsTest, TestProjectUpdatesReorderedColumns) {
    Schema client_schema({ ColumnSchema("key", INT32),
                           ColumnSchema("string_val", STRING, true),
                           ColumnSchema("int_val", INT32) },
                         1);
    Schema server_schema({ ColumnSchema("key", INT32),
                           ColumnSchema("int_val", INT32),
                           ColumnSchema("string_val", STRING, true) },
                         1);
    server_schema = SchemaBuilder(server_schema).Build();
  
    KuduPartialRow client_row(&client_schema);
    ASSERT_OK(client_row.SetInt32("key", 12345));
    ASSERT_OK(client_row.SetInt32("int_val", 54321));
    EXPECT_EQ("MUTATE (int32 key=12345) SET int_val=54321",
              TestProjection(RowOperationsPB::UPDATE, client_row, server_schema));
  }
  
  // Client schema is missing one of the columns in the server schema.
  // This is OK on an update.
  TEST_F(RowOperationsTest, DISABLED_TestProjectUpdatesSubsetOfColumns) {
    Schema client_schema({ ColumnSchema("key", INT32),
                           ColumnSchema("string_val", STRING, true) },
                         1);
    Schema server_schema({ ColumnSchema("key", INT32),
                           ColumnSchema("int_val", INT32),
                           ColumnSchema("string_val", STRING, true) },
                         1);
    server_schema = SchemaBuilder(server_schema).Build();
  
    KuduPartialRow client_row(&client_schema);
    ASSERT_OK(client_row.SetInt32("key", 12345));
    ASSERT_OK(client_row.SetStringNoCopy("string_val", "foo"));
    EXPECT_EQ("MUTATE (int32 key=12345) SET string_val=foo",
              TestProjection(RowOperationsPB::UPDATE, client_row, server_schema));
  }
  
  TEST_F(RowOperationsTest, TestClientMismatchedType) {
    Schema client_schema({ ColumnSchema("key", INT32),
                           ColumnSchema("int_val", INT8) },
                         1);
    Schema server_schema({ ColumnSchema("key", INT32),
                           ColumnSchema("int_val", INT32) },
                         1);
    server_schema = SchemaBuilder(server_schema).Build();
  
    KuduPartialRow client_row(&client_schema);
    ASSERT_OK(client_row.SetInt32("key", 12345));
    ASSERT_OK(client_row.SetInt8("int_val", 1));
    EXPECT_EQ("error: Invalid argument: The column 'int_val' must have type "
              "INT32 NOT NULL found INT8 NOT NULL",
              TestProjection(RowOperationsPB::UPDATE, client_row, server_schema));
  }
  
  TEST_F(RowOperationsTest, TestProjectDeletes) {
    Schema client_schema({ ColumnSchema("key", INT32),
                           ColumnSchema("key_2", INT32),
                           ColumnSchema("int_val", INT32),
                           ColumnSchema("string_val", STRING, true) },
                         2);
    Schema server_schema = SchemaBuilder(client_schema).Build();
  
    KuduPartialRow client_row(&client_schema);
    // No columns set
    EXPECT_EQ("row error: Invalid argument: No value provided for key column: key INT32 NOT NULL",
              TestProjection(RowOperationsPB::DELETE, client_row, server_schema));
  
    // Only half the key set
    ASSERT_OK(client_row.SetInt32("key", 12345));
    EXPECT_EQ("row error: Invalid argument: No value provided for key column: key_2 INT32 NOT NULL",
              TestProjection(RowOperationsPB::DELETE, client_row, server_schema));
  
    // Whole key set (correct)
    ASSERT_OK(client_row.SetInt32("key_2", 54321));
    EXPECT_EQ("MUTATE (int32 key=12345, int32 key_2=54321) DELETE",
              TestProjection(RowOperationsPB::DELETE, client_row, server_schema));
  
    // Extra column set
    ASSERT_OK(client_row.SetNull("string_val"));
    EXPECT_EQ("row error: Invalid argument: DELETE should not have a value for column: "
              "string_val STRING NULLABLE",
              TestProjection(RowOperationsPB::DELETE, client_row, server_schema));
  
    // Extra column set (incorrect)
    ASSERT_OK(client_row.SetStringNoCopy("string_val", "hello"));
    EXPECT_EQ("row error: Invalid argument: DELETE should not have a value for column: "
              "string_val STRING NULLABLE",
              TestProjection(RowOperationsPB::DELETE, client_row, server_schema));
  
    // Force to set null on key column.
    {
      KuduPartialRow client_row(&client_schema);
      int col_idx;
      ASSERT_OK(client_schema.FindColumn("key", &col_idx));
      ContiguousRow row(&client_schema, client_row.row_data_);
      row.set_null(col_idx, true);
      BitmapSet(client_row.isset_bitmap_, col_idx);
      EXPECT_EQ("row error: Invalid argument: NULL values not allowed for key column: "
                "key INT32 NOT NULL",
                TestProjection(RowOperationsPB::DELETE, client_row, server_schema));
    }
  
    // Force to set null on non-key column.
    {
      KuduPartialRow client_row(&client_schema);
      ASSERT_OK(client_row.SetInt32("key", 12345));
      ASSERT_OK(client_row.SetInt32("key_2", 12345));
      int col_idx;
      ASSERT_OK(client_schema.FindColumn("int_val", &col_idx));
      ContiguousRow row(&client_schema, client_row.row_data_);
      row.set_null(col_idx, true);
      BitmapSet(client_row.isset_bitmap_, col_idx);
      EXPECT_EQ("row error: Invalid argument: DELETE should not have a value for column: "
                "int_val INT32 NOT NULL",
                TestProjection(RowOperationsPB::DELETE, client_row, server_schema));
    }
  }
  
  TEST_F(RowOperationsTest, SplitKeyRoundTrip) {
    Schema client_schema = Schema({ ColumnSchema("int8", INT8),
                                    ColumnSchema("int16", INT16),
                                    ColumnSchema("int32", INT32),
                                    ColumnSchema("int64", INT64),
                                    ColumnSchema("string", STRING),
                                    ColumnSchema("binary", BINARY),
                                    ColumnSchema("timestamp", UNIXTIME_MICROS),
                                    ColumnSchema("missing", STRING) },
                                  8);
  
    // Use values at the upper end of the range.
    int8_t int8_expected = 0xFE;
    int16_t int16_expected = 0xFFFE;
    int32_t int32_expected = 0xFFFFFE;
    int64_t int64_expected = 0xFFFFFFFE;
  
    KuduPartialRow row(&client_schema);
    ASSERT_OK(row.SetInt8("int8", int8_expected));
    ASSERT_OK(row.SetInt16("int16", int16_expected));
    ASSERT_OK(row.SetInt32("int32", int32_expected));
    ASSERT_OK(row.SetInt64("int64", int64_expected));
    ASSERT_OK(row.SetStringNoCopy("string", "string-value"));
    ASSERT_OK(row.SetBinaryNoCopy("binary", "binary-value"));
    ASSERT_OK(row.SetUnixTimeMicros("timestamp", 9));
  
    RowOperationsPB pb;
    RowOperationsPBEncoder(&pb).Add(RowOperationsPB::SPLIT_ROW, row);
  
    Schema schema = client_schema.CopyWithColumnIds();
    RowOperationsPBDecoder decoder(&pb, &client_schema, &schema, nullptr);
    vector<DecodedRowOperation> ops;
    ASSERT_OK(decoder.DecodeOperations<DecoderMode::SPLIT_ROWS>(&ops));
    ASSERT_EQ(1, ops.size());
  
    const shared_ptr<KuduPartialRow>& row2 = ops[0].split_row;
  
    int8_t int8_val;
    ASSERT_OK(row2->GetInt8("int8", &int8_val));
    CHECK_EQ(int8_expected, int8_val);
  
    int16_t int16_val;
    ASSERT_OK(row2->GetInt16("int16", &int16_val));
    CHECK_EQ(int16_expected, int16_val);
  
    int32_t int32_val;
    ASSERT_OK(row2->GetInt32("int32", &int32_val));
    CHECK_EQ(int32_expected, int32_val);
  
    int64_t int64_val;
    ASSERT_OK(row2->GetInt64("int64", &int64_val));
    CHECK_EQ(int64_expected, int64_val);
  
    Slice string_val;
    ASSERT_OK(row2->GetString("string", &string_val));
    CHECK_EQ("string-value", string_val);
  
    Slice binary_val;
    ASSERT_OK(row2->GetBinary("binary", &binary_val));
    CHECK_EQ(Slice("binary-value"), binary_val);
  
    CHECK(!row2->IsColumnSet("missing"));
  }
  
  namespace {
  void CheckExceedCellLimit(const Schema& client_schema,
                            const vector<string>& col_values,
                            RowOperationsPB::Type op_type,
                            const Status& expect_status,
                            const string& expect_msg) {
    ASSERT_EQ(client_schema.num_columns(), col_values.size());
  
    // Fill the row.
    KuduPartialRow row(&client_schema);
    for (size_t i = 0; i < client_schema.num_columns(); ++i) {
      if (op_type == RowOperationsPB::DELETE && i >= client_schema.num_key_columns()) {
        // DELETE should not have a value for non-key column.
        break;
      }
      const ColumnSchema& column_schema = client_schema.column(i);
      switch (column_schema.type_info()->type()) {
        case STRING:
          ASSERT_OK(row.SetStringNoCopy(static_cast<int>(i), col_values[i]));
          break;
        case BINARY:
          ASSERT_OK(row.SetBinaryNoCopy(static_cast<int>(i), col_values[i]));
          break;
        default:
          ASSERT_TRUE(false) << "Unsupported type: " << column_schema.ToString();
      }
    }
  
    RowOperationsPB pb;
    RowOperationsPBEncoder(&pb).Add(op_type, row);
  
    Arena arena(1024*1024);
    Schema schema = client_schema.CopyWithColumnIds();
    RowOperationsPBDecoder decoder(&pb, &client_schema, &schema, &arena);
    vector<DecodedRowOperation> ops;
    Status s;
    switch (op_type) {
      case RowOperationsPB::INSERT:
      case RowOperationsPB::UPDATE:
      case RowOperationsPB::DELETE:
      case RowOperationsPB::UPSERT:
        s = decoder.DecodeOperations<WRITE_OPS>(&ops);
        break;
      case RowOperationsPB::SPLIT_ROW:
      case RowOperationsPB::RANGE_LOWER_BOUND:
      case RowOperationsPB::RANGE_UPPER_BOUND:
      case RowOperationsPB::EXCLUSIVE_RANGE_LOWER_BOUND:
      case RowOperationsPB::INCLUSIVE_RANGE_UPPER_BOUND:
        s = decoder.DecodeOperations<SPLIT_ROWS>(&ops);
        break;
      default:
        ASSERT_TRUE(false) << "Unsupported op_type " << op_type;
    }
    ASSERT_OK(s);
    for (const auto& op : ops) {
      ASSERT_EQ(op.result.CodeAsString(), expect_status.CodeAsString());
      ASSERT_STR_CONTAINS(op.result.ToString(), expect_msg);
    }
  }
  
  void CheckInsertUpsertExceedCellLimit(const Schema& client_schema,
                                        const vector<string>& col_values,
                                        const Status& expect_status,
                                        const string& expect_msg) {
    for (auto op_type : { RowOperationsPB::INSERT,
                          RowOperationsPB::UPSERT }) {
      NO_FATALS(CheckExceedCellLimit(client_schema, col_values, op_type, expect_status, expect_msg));
    }
  }
  
  void CheckUpdateDeleteExceedCellLimit(const Schema& client_schema,
                                        const vector<string>& col_values,
                                        const Status& expect_status,
                                        const string& expect_msg) {
    for (auto op_type : { RowOperationsPB::UPDATE,
                          RowOperationsPB::DELETE }) {
      NO_FATALS(CheckExceedCellLimit(client_schema, col_values, op_type, expect_status, expect_msg));
    }
  }
  
  void CheckWriteExceedCellLimit(const Schema& client_schema,
                                 const vector<string>& col_values,
                                 const Status& expect_status,
                                 const string& expect_msg) {
    NO_FATALS(CheckInsertUpsertExceedCellLimit(client_schema, col_values, expect_status, expect_msg));
    NO_FATALS(CheckUpdateDeleteExceedCellLimit(client_schema, col_values, expect_status, expect_msg));
  }
  
  void CheckSplitExceedCellLimit(const Schema& client_schema,
                                 const vector<string>& col_values,
                                 const Status& expect_status,
                                 const string& expect_msg) {
    for (auto op_type : { RowOperationsPB::SPLIT_ROW,
                          RowOperationsPB::RANGE_LOWER_BOUND,
                          RowOperationsPB::RANGE_UPPER_BOUND,
                          RowOperationsPB::EXCLUSIVE_RANGE_LOWER_BOUND,
                          RowOperationsPB::INCLUSIVE_RANGE_UPPER_BOUND }) {
      NO_FATALS(CheckExceedCellLimit(client_schema, col_values, op_type, expect_status, expect_msg));
    }
  }
  } // anonymous namespace
  
  TEST_F(RowOperationsTest, ExceedCellLimit) {
    Schema client_schema = Schema({ ColumnSchema("key_string", STRING),
                                    ColumnSchema("key_binary", BINARY),
                                    ColumnSchema("col_string", STRING),
                                    ColumnSchema("col_binary", BINARY) },
                                  2);
  
    const string long_string(static_cast<size_t>(FLAGS_max_cell_size_bytes), 'a');
    const string too_long_string(static_cast<size_t>(FLAGS_max_cell_size_bytes + 1), 'a');
    const vector<string> base_col_values(4, long_string);
  
    // All column cell sizes are not exceed.
    NO_FATALS(CheckWriteExceedCellLimit(client_schema, base_col_values, Status::OK(), ""));
  
    // Some column cell size exceed for INSERT and UPSERT operation.
    for (size_t i = 0; i < client_schema.num_columns(); ++i) {
      auto col_values(base_col_values);
      col_values[i] = too_long_string;
      NO_FATALS(CheckInsertUpsertExceedCellLimit(client_schema,
                                                 col_values,
                                                 Status::InvalidArgument(""),
                                                 Substitute("value too large for column '$0'"
                                                            " ($1 bytes, maximum is $2 bytes)",
                                                            client_schema.column(i).name(),
                                                            FLAGS_max_cell_size_bytes + 1,
                                                            FLAGS_max_cell_size_bytes)));
    }
  
    // Key column cell size exceed for UPDATE and DELETE operation, it's OK.
    for (size_t i = 0; i < client_schema.num_key_columns(); ++i) {
      auto col_values(base_col_values);
      col_values[i] = too_long_string;
      NO_FATALS(CheckUpdateDeleteExceedCellLimit(client_schema, col_values, Status::OK(), ""));
    }
  
    // Non-key column cell size exceed for UPDATE operation.
    for (size_t i = client_schema.num_key_columns(); i < client_schema.num_columns(); ++i) {
      auto col_values(base_col_values);
      col_values[i] = too_long_string;
      NO_FATALS(CheckExceedCellLimit(client_schema,
                                     col_values,
                                     RowOperationsPB::UPDATE,
                                     Status::InvalidArgument(""),
                                     Substitute("value too large for column '$0'"
                                                " ($1 bytes, maximum is $2 bytes)",
                                                client_schema.column(i).name(),
                                                FLAGS_max_cell_size_bytes + 1,
                                                FLAGS_max_cell_size_bytes)));
    }
  
    // Some column cell size exceed for SPLIT_ROW type operation, it's OK.
    for (size_t i = 0; i < client_schema.num_columns(); ++i) {
      auto col_values(base_col_values);
      col_values[i] = too_long_string;
      NO_FATALS(CheckSplitExceedCellLimit(client_schema, col_values, Status::OK(), ""));
    }
  }
  
  } // namespace kudu
  