YugabyteDB (2.13.0.0-b42, bfc6a6643e7399ac8a0e81d06a3ee6d6571b33ab)

// Copyright (c) YugaByte, Inc.

//

// Licensed 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 "yb/yql/pgwrapper/pg_mini_test_base.h"

#include "yb/util/test_macros.h"

#include "yb/util/test_thread_holder.h"

using namespace std::literals;

DECLARE_bool(TEST_fail_in_apply_if_no_metadata);

namespace yb {

namespace pgwrapper {

class PgTxnTest : public PgMiniTestBase {

};

TEST_F(PgTxnTest, YB_DISABLE_TEST_IN_SANITIZERS(EmptyUpdate)) {

  FLAGS_TEST_fail_in_apply_if_no_metadata = true;

  auto conn = ASSERT_RESULT(Connect());

  ASSERT_OK(conn.Execute("CREATE TABLE test (key TEXT, value TEXT, PRIMARY KEY((key) HASH))"));

  ASSERT_OK(conn.StartTransaction(IsolationLevel::SNAPSHOT_ISOLATION));

  ASSERT_OK(conn.Execute("UPDATE test SET value = 'a' WHERE key = 'b'"));

  ASSERT_OK(conn.CommitTransaction());

}

class PgTxnRF1Test : public PgTxnTest {

 public:

  size_t NumTabletServers() override {

    return 1;

  }

};

TEST_F_EX(PgTxnTest, YB_DISABLE_TEST_IN_TSAN(SelectRF1ReadOnlyDeferred), PgTxnRF1Test) {

  auto conn = ASSERT_RESULT(Connect());

  ASSERT_OK(conn.Execute("CREATE TABLE test (key INT)"));

  ASSERT_OK(conn.Execute("INSERT INTO test VALUES (1)"));

  ASSERT_OK(conn.Execute("BEGIN ISOLATION LEVEL SERIALIZABLE, READ ONLY, DEFERRABLE"));

  auto res = ASSERT_RESULT(conn.FetchValue<int32_t>("SELECT * FROM test"));

  ASSERT_EQ(res, 1);

  ASSERT_OK(conn.Execute("COMMIT"));

}

TEST_F(PgTxnTest, YB_DISABLE_TEST_IN_TSAN(SerializableReadWriteConflicts)) {

  auto conn1 = ASSERT_RESULT(Connect());

  auto conn2 = ASSERT_RESULT(Connect());

  constexpr double kPriorityBound = 0.5;

  ASSERT_OK(conn1.ExecuteFormat("SET yb_transaction_priority_lower_bound = $0", kPriorityBound));

  ASSERT_OK(conn1.Execute("CREATE TABLE test (key INT, value INT, PRIMARY KEY((key) HASH))"));

  ASSERT_OK(conn1.Execute("CREATE INDEX idx ON test (value)"));

  ASSERT_OK(conn1.StartTransaction(IsolationLevel::SERIALIZABLE_ISOLATION));

  ASSERT_OK(conn1.Execute("INSERT INTO test VALUES (1, 1)"));

  ASSERT_OK(conn2.ExecuteFormat("SET yb_transaction_priority_upper_bound = $0", kPriorityBound));

  ASSERT_OK(conn2.StartTransaction(IsolationLevel::SERIALIZABLE_ISOLATION));

  ASSERT_NOK(conn2.Fetch("SELECT key FROM test WHERE value = 1"));

  ASSERT_OK(conn1.Execute("COMMIT"));

}

// Test concurrently insert increasing values, and in parallel perform read of several recent

// values.

// Checking that reads could be serialized.

TEST_F(PgTxnTest, YB_DISABLE_TEST_IN_TSAN(ReadRecentSet)) {

  auto conn = ASSERT_RESULT(Connect());

  constexpr int kWriters = 16;

  constexpr int kReaders = 16;

  constexpr int kReadLength = 32;

  ASSERT_OK(conn.Execute(

      "CREATE TABLE test (key INT, value INT, PRIMARY KEY((key) HASH)) SPLIT INTO 1 TABLETS"));

  ASSERT_OK(conn.Execute(

      "CREATE INDEX idx ON test (value) SPLIT INTO 2 TABLETS"));

  TestThreadHolder thread_holder;

  std::atomic<int> value(0);

  for (int i = 0; i != kWriters; ++i) {

    thread_holder.AddThreadFunctor(

        [this, &stop = thread_holder.stop_flag(), &value] {

      auto connection = ASSERT_RESULT(Connect());

      while (!stop.load()) {

        int cur = value.fetch_add(1);

        ASSERT_OK(connection.StartTransaction(IsolationLevel::SERIALIZABLE_ISOLATION));

        auto status = connection.ExecuteFormat("INSERT INTO test VALUES ($0, $0)", cur);

        if (status.ok()) {

          status = connection.CommitTransaction();

        }

        if (!status.ok()) {

          ASSERT_OK(connection.RollbackTransaction());

        }

      }

    });

  }

  std::mutex reads_mutex;

  struct Read {

    int read_min;

    uint64_t mask;

    static std::string ValuesToString(int read_min, uint64_t mask) {

      auto v = read_min;

      auto m = mask;

      std::vector<int> values;

      while (m) {

        if (m & 1ULL) {

          values.push_back(v);

        }

        ++v;

        m >>= 1ULL;

      }

      return AsString(values);

    }

    std::string ToString() const {

      return Format("{ read range: $0-$1 values: $2 }",

                    read_min, read_min + kReadLength - 1, ValuesToString(read_min, mask));

    }

  };

  std::vector<Read> reads;

  for (int i = 0; i != kReaders; ++i) {

    thread_holder.AddThreadFunctor(

        [this, &stop = thread_holder.stop_flag(), &value, &reads, &reads_mutex] {

      auto connection = ASSERT_RESULT(Connect());

      char str_buffer[0x200];

      while (!stop.load()) {

        const auto read_min = std::max(value.load() - kReadLength, 0);

        char* p = str_buffer;

        for (auto v = read_min; v != read_min + kReadLength; ++v) {

          if (p != str_buffer) {

            *p++ = ',';

          }

          p = FastInt64ToBufferLeft(v, p);

        }

        ASSERT_OK(connection.StartTransaction(IsolationLevel::SERIALIZABLE_ISOLATION));

        auto res = connection.FetchFormat("SELECT value FROM test WHERE value in ($0)", str_buffer);

        if (!res.ok()) {

          ASSERT_OK(connection.RollbackTransaction());

          continue;

        }

        auto status = connection.CommitTransaction();

        if (!status.ok()) {

          ASSERT_OK(connection.RollbackTransaction());

          continue;

        }

        uint64_t mask = 0;

        for (int j = 0, count = PQntuples(res->get()); j != count; ++j) {

          mask |= 1ULL << (ASSERT_RESULT(GetInt32(res->get(), j, 0)) - read_min);

        }

        std::lock_guard<std::mutex> lock(reads_mutex);

        Read new_read{read_min, mask};

        reads.erase(std::remove_if(reads.begin(), reads.end(),

            [&new_read, &stop](const auto& old_read) {

          int read_min_delta = new_read.read_min - old_read.read_min;

          // Existing read is too old, remove it.

          if (read_min_delta >= 64) {

            return true;

          }

          // New read is too old, cannot check it.

          if (read_min_delta <= -64) {

            return false;

          }

          constexpr auto kFullMask = (1ULL << kReadLength) - 1ULL;

          // Extract only numbers that belong to both reads.

          uint64_t lmask, rmask;

          if (read_min_delta >= 0) {

            lmask = new_read.mask & (kFullMask >> read_min_delta);

            rmask = (old_read.mask >> read_min_delta) & kFullMask;

          } else {

            lmask = (new_read.mask >> -read_min_delta) & kFullMask;

            rmask = old_read.mask & (kFullMask >> -read_min_delta);

          }

          // Check that one set is subset of another subset.

          // I.e. only one set is allowed to have elements that is not contained in another set.

          if ((lmask | rmask) != std::max(lmask, rmask)) {

            const auto read = std::max(old_read.read_min, new_read.read_min);

            ADD_FAILURE() << "R1: " << old_read.ToString() << "\nR2: " << new_read.ToString()

                          << "\nR1-R2: " << Read::ValuesToString(read, rmask ^ (lmask & rmask))

                          << ", R2-R1: " << Read::ValuesToString(read, lmask ^ (lmask & rmask));

            stop.store(true);

          }

          return false;

        }), reads.end());

        reads.push_back(new_read);

      }

    });

  }

  thread_holder.WaitAndStop(30s);

}

} // namespace pgwrapper

} // namespace yb