<commit_before>package com.datastax.spark.connector.cql

import com.datastax.oss.driver.api.core.config.DefaultDriverOption
import com.datastax.oss.driver.api.core.cql.SimpleStatement
import com.datastax.spark.connector.cluster.DefaultCluster
import com.datastax.spark.connector.{SparkCassandraITFlatSpecBase, _}
import org.scalatest.BeforeAndAfterEach

import scala.collection.JavaConverters._

case class KeyValue(key: Int, group: Long, value: String)
case class KeyValueWithConversion(key: String, group: Int, value: Long)

class CassandraConnectorSpec extends SparkCassandraITFlatSpecBase with DefaultCluster with BeforeAndAfterEach{

  override lazy val conn = CassandraConnector(defaultConf)

  val createKeyspaceCql = keyspaceCql(ks)

  "A CassandraConnector" should "connect to Cassandra with native protocol" in {
    conn.withSessionDo { session =>
      assert(session.isClosed === false)
      assert(session !== null)
    }
  }

  it should "give access to cluster metadata" in {
    conn.withSessionDo { session =>
      assert(session.getMetadata.getNodes.size > 0)
    }
  }

  it should "have the default max hosts in pooling options" in {
    val conf = conn.withSessionDo(_.getContext.getConfig.getDefaultProfile)
    conf.getInt(DefaultDriverOption.CONNECTION_POOL_LOCAL_SIZE) should be >= 1
    conf.getInt(DefaultDriverOption.CONNECTION_POOL_REMOTE_SIZE) should be >= 1
  }

  it should "have larger max hosts if set" in {
    val alteredConnector = CassandraConnector(
      defaultConf
          .set(CassandraConnectorConf.LocalConnectionsPerExecutorParam.name, "5")
          .set(CassandraConnectorConf.RemoteConnectionsPerExecutorParam.name, "3"))

    // *ConnectionsPerExecutorsParam are not taken in account when retrieving session objects from global cache.
    // This results in a possibility of grabbing a session that does not have the parameters set.
    // https://github.com/riptano/bdp/pull/11359/files#diff-a866d6dfb859aa080a01d9a55ae1e5c0R43
    CassandraConnector.evictCache()

    val conf = alteredConnector.withSessionDo(_.getContext.getConfig.getDefaultProfile)
    conf.getInt(DefaultDriverOption.CONNECTION_POOL_LOCAL_SIZE) should be (5)
    conf.getInt(DefaultDriverOption.CONNECTION_POOL_REMOTE_SIZE) should be (3)
  }

  it should "run queries" in {
    conn.withSessionDo { session =>
      session.execute(createKeyspaceCql)
      session.execute(s"DROP TABLE IF EXISTS $ks.simple_query")
      session.execute(s"CREATE TABLE $ks.simple_query (key INT PRIMARY KEY, value TEXT)")
      session.execute(s"INSERT INTO $ks.simple_query(key, value) VALUES (1, 'value')")
      val result = session.execute(
        SimpleStatement.newInstance(s"SELECT * FROM $ks.simple_query WHERE key = 1"))

      assert(result.one().getString("value") === "value")
    }
  }

  it should "disconnect from the session after use" in {
    val session = conn.withSessionDo { session => session }
    Thread.sleep(
      sc.getConf.getInt(
        "spark.cassandra.connection.keepAliveMS",
        CassandraConnectorConf.KeepAliveMillisParam.default) * 2)
    assert(session.isClosed === true)
  }

  it should "share internal Cluster and Session object between multiple logical sessions" in {
    val session1 = conn.openSession()
    val threadCount1 = Thread.activeCount()
    val session2 = conn.openSession()
    val threadCount2 = Thread.activeCount()

    // Unfortunately we don't have a way to obtain an internal Session object; all we got here are proxies.
    // Instead, we try to figure out whether a new Session was opened by counting active threads.
    // Opening internal Session creates new threads, so if none was created, the thread count would not change.
    threadCount1 shouldEqual threadCount2
    session1.close()
    session1.isClosed shouldEqual true
    session2.isClosed shouldEqual false
    session2.close()
    session2.isClosed shouldEqual true
  }

  it should "share internal Cluster object between multiple logical sessions created by different connectors to the same cluster" in {
    val conn2 = CassandraConnector(sc.getConf)
    val session1 = conn.openSession()
    val threadCount1 = Thread.activeCount()
    val session2 = conn2.openSession()
    val threadCount2 = Thread.activeCount()
    threadCount1 shouldEqual threadCount2
    session1.close()
    session1.isClosed shouldEqual true
    session2.isClosed shouldEqual false
    session2.close()
    session2.isClosed shouldEqual true
  }

  it should "cache session objects for reuse" in {
    CassandraConnector(sc.getConf).withSessionDo(x => {})
    CassandraConnector(sc.getConf).withSessionDo(x => {})
    val sessionCache = CassandraConnector.sessionCache
    sessionCache.contains(CassandraConnectorConf(sc.getConf)) should be (true)
    sessionCache.cache.size should be (1)
  }<commit_msg>DSP-18857 add session name comparison<commit_after>package com.datastax.spark.connector.cql

import com.datastax.oss.driver.api.core.config.DefaultDriverOption
import com.datastax.oss.driver.api.core.cql.SimpleStatement
import com.datastax.spark.connector.cluster.DefaultCluster
import com.datastax.spark.connector.{SparkCassandraITFlatSpecBase, _}
import org.scalatest.BeforeAndAfterEach

import scala.collection.JavaConverters._

case class KeyValue(key: Int, group: Long, value: String)
case class KeyValueWithConversion(key: String, group: Int, value: Long)

class CassandraConnectorSpec extends SparkCassandraITFlatSpecBase with DefaultCluster with BeforeAndAfterEach{

  override lazy val conn = CassandraConnector(defaultConf)

  val createKeyspaceCql = keyspaceCql(ks)

  "A CassandraConnector" should "connect to Cassandra with native protocol" in {
    conn.withSessionDo { session =>
      assert(session.isClosed === false)
      assert(session !== null)
    }
  }

  it should "give access to cluster metadata" in {
    conn.withSessionDo { session =>
      assert(session.getMetadata.getNodes.size > 0)
    }
  }

  it should "have the default max hosts in pooling options" in {
    val conf = conn.withSessionDo(_.getContext.getConfig.getDefaultProfile)
    conf.getInt(DefaultDriverOption.CONNECTION_POOL_LOCAL_SIZE) should be >= 1
    conf.getInt(DefaultDriverOption.CONNECTION_POOL_REMOTE_SIZE) should be >= 1
  }

  it should "have larger max hosts if set" in {
    val alteredConnector = CassandraConnector(
      defaultConf
          .set(CassandraConnectorConf.LocalConnectionsPerExecutorParam.name, "5")
          .set(CassandraConnectorConf.RemoteConnectionsPerExecutorParam.name, "3"))

    // *ConnectionsPerExecutorsParam are not taken in account when retrieving session objects from global cache.
    // This results in a possibility of grabbing a session that does not have the parameters set.
    // https://github.com/riptano/bdp/pull/11359/files#diff-a866d6dfb859aa080a01d9a55ae1e5c0R43
    CassandraConnector.evictCache()

    val conf = alteredConnector.withSessionDo(_.getContext.getConfig.getDefaultProfile)
    conf.getInt(DefaultDriverOption.CONNECTION_POOL_LOCAL_SIZE) should be (5)
    conf.getInt(DefaultDriverOption.CONNECTION_POOL_REMOTE_SIZE) should be (3)
  }

  it should "run queries" in {
    conn.withSessionDo { session =>
      session.execute(createKeyspaceCql)
      session.execute(s"DROP TABLE IF EXISTS $ks.simple_query")
      session.execute(s"CREATE TABLE $ks.simple_query (key INT PRIMARY KEY, value TEXT)")
      session.execute(s"INSERT INTO $ks.simple_query(key, value) VALUES (1, 'value')")
      val result = session.execute(
        SimpleStatement.newInstance(s"SELECT * FROM $ks.simple_query WHERE key = 1"))

      assert(result.one().getString("value") === "value")
    }
  }

  it should "disconnect from the session after use" in {
    val session = conn.withSessionDo { session => session }
    Thread.sleep(
      sc.getConf.getInt(
        "spark.cassandra.connection.keepAliveMS",
        CassandraConnectorConf.KeepAliveMillisParam.default) * 2)
    assert(session.isClosed === true)
  }

  it should "share internal Cluster and Session object between multiple logical sessions" in {
    val session1 = conn.openSession()
    val threadCount1 = Thread.activeCount()
    val session2 = conn.openSession()
    val threadCount2 = Thread.activeCount()

    session1.getName shouldEqual session2.getName
    // Unfortunately we don't have a way to obtain an internal Session object; all we got here are proxies.
    // Instead, we try to figure out whether a new Session was opened by counting active threads.
    // Opening internal Session creates new threads, so if none was created, the thread count would not change.
    threadCount1 shouldEqual threadCount2
    session1.close()
    session1.isClosed shouldEqual true
    session2.isClosed shouldEqual false
    session2.close()
    session2.isClosed shouldEqual true
  }

  it should "share internal Cluster object between multiple logical sessions created by different connectors to the same cluster" in {
    val conn2 = CassandraConnector(sc.getConf)
    val session1 = conn.openSession()
    val threadCount1 = Thread.activeCount()
    val session2 = conn2.openSession()
    val threadCount2 = Thread.activeCount()
    threadCount1 shouldEqual threadCount2

    session1.getName shouldEqual session2.getName
    session1.close()
    session1.isClosed shouldEqual true
    session2.isClosed shouldEqual false
    session2.close()
    session2.isClosed shouldEqual true
  }

  it should "cache session objects for reuse" in {
    CassandraConnector(sc.getConf).withSessionDo(x => {})
    CassandraConnector(sc.getConf).withSessionDo(x => {})
    val sessionCache = CassandraConnector.sessionCache
    sessionCache.contains(CassandraConnectorConf(sc.getConf)) should be (true)
    sessionCache.cache.size should be (1)
  }
{'fraction_non_alphanumeric': 0.08271137920458031, 'fraction_numerical': 0.013188835497392904, 'mean_word_length': 4.108093994778068, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 21, 'https://': 2, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '833984', 'n_tokens_mistral': 2873, 'n_tokens_neox': 2783, 'n_words': 760}