//设置package包名称以及导入依赖的类
import java.sql.{Driver, DriverManager}

import org.scalatest.BeforeAndAfter
import org.scalatestplus.play.PlaySpec
import play.api.Logger
import scalikejdbc._
import scalikejdbc.config.DBs

import scala.collection.mutable.Stack

class DatabaseSpec extends PlaySpec with BeforeAndAfter {
  implicit var session = AutoSession

  before {
    DBs.setupAll()
  }

  after {
    DBs.closeAll()
  }

  "Database" must {
    "List of database drivers" in {
      val drivers = DriverManager.getDrivers

      while (drivers.hasMoreElements) {
        val driver:Driver = drivers.nextElement()
        Logger.info(driver.toString)
      }
    }
    "Fetch users" in {
      val sql: SQL[Nothing, NoExtractor] = sql"select * from users"
      val map: SQL[Map[String, Any], HasExtractor] = sql.toMap
      val list: SQLToList[Map[String, Any], HasExtractor] = map.list
      val users: List[Map[String, Any]] = list.apply()
    }
  }

  "Slick" must {
    "How to use Scala test" in {
      // TODO
    }

    "throw NoSuchElementException if an empty stack is popped" in {
      val emptyStack = new Stack[Int]
      a [NoSuchElementException] must be thrownBy {
        emptyStack.pop()
      }
    }
  }
} 

//设置package包名称以及导入依赖的类
package com.unity.analytics.spark.utils.parquet

import java.sql.DriverManager

import org.apache.spark.Logging
import org.json4s.jackson.Serialization


case class BackupMetadata(
                           backupId: String,
                           backupEntries: Array[BackupEntry]
                         )

case class BackupEntry(
                        srcDir: String,
                        destDir: String,
                        srcNumFiles: Int = 1,
                        destNumFiles: Int = 1
                      )

object BackupMetadata extends Logging {
  val tableName = "backup_metadata"
  implicit val formats = org.json4s.DefaultFormats

  def write(backupId: String, backupEntries: Array[BackupEntry], jdbcConfig: Map[String, String]): Unit = {
    val connection = DriverManager.getConnection(jdbcConfig.get("url").get)
    val backupEntriesJSON = Serialization.write[Array[BackupEntry]](backupEntries)
    val sql = s"""INSERT INTO ${BackupMetadata.tableName} (id, entries) VALUES ('$backupId', '$backupEntriesJSON') ON DUPLICATE KEY UPDATE entries = '$backupEntriesJSON'"""

    try {
      connection.prepareStatement(sql).execute()
    }
    finally {
      connection.close()
    }
  }

  def read(backupId: String, jdbcConfig: Map[String, String]): Option[BackupMetadata] = {
    //Read from MySQL
    val connection = DriverManager.getConnection(jdbcConfig.get("url").get)
    val sql = s"SELECT * FROM $tableName WHERE id = '$backupId'"
    try {
      val results = connection.prepareStatement(sql).executeQuery()
      while (results.next()) {
        val backupEntriesJSON = results.getString("entries")
        val backupEntries = Serialization.read[Array[BackupEntry]](backupEntriesJSON)
        return Some(BackupMetadata(backupId, backupEntries))
      }
    }
    catch {
      case e: Exception => {
        logError(s"Error loading backup BackupMetadata $backupId - ${e.getMessage}")
      }
    }
    finally {
      connection.close()
    }
    None
  }
} 

//设置package包名称以及导入依赖的类
package com.unity.analytics.spark.utils.parquet

import java.io.{File, StringReader}
import java.sql.DriverManager

import org.apache.commons.io.FileUtils
import org.h2.tools.{RunScript, Server}

class BackupMetadataTest extends BaseTest {

  val dbDriver = "org.h2.Driver"
  val server = Server.createTcpServer("-tcpPort", "9999")
  val jdbcDB = "dummydb"
  val jdbcUrl = s"jdbc:h2:mem:$jdbcDB;DATABASE_TO_UPPER=FALSE;MODE=MYSQL"
  val jdbcConfig = Map(
    "url" -> jdbcUrl
  )
  val backendConnection = DriverManager.getConnection(jdbcUrl)

  val backupId = "dummyBackupId"

  override def beforeAll(): Unit = {
    super.beforeAll()
    println("INITIALIZE H2")
    // Initialize H2
    val reader = new StringReader(FileUtils.readFileToString(new File("src/test/scala/resources/backup_metadata.sql")).replace('`', '"'))
    RunScript.execute(backendConnection, reader)
  }

  test("Test reading and writing backup metadata") {
    val entries = Array(
      BackupEntry("hdfs://hello", "s3a://hello", 100, 10),
      BackupEntry("/source", "/destination", 200, 2)
    ).sortBy(_.destDir)
    BackupMetadata.write(backupId, entries, jdbcConfig)
    val metadata = BackupMetadata.read(backupId, jdbcConfig).get
    assert(metadata.backupId === backupId)
    assert(metadata.backupEntries.length === entries.length)
    val output = metadata.backupEntries.sortBy(_.destDir)
    entries.zip(output).foreach(e => {
      assert(e._1 === e._2)
    })
  }
} 

//设置package包名称以及导入依赖的类
package com.dao

import java.sql.{Connection, DriverManager, SQLException}


object OracleUtil {
  val driverClass = "oracle.jdbc.driver.OracleDriver"
  val jdbcUrl = "jdbc:oracle:thin:@(DESCRIPTION=(ADDRESS_LIST=(ADDRESS=(PROTOCOL=TCP)(HOST=172.30.12.26)(PORT=1521))(FAILOVER=ON)(LOAD_BALANCE=ON))(CONNECT_DATA= (SERVICE_NAME=msds)))"
  val username = "acctpaymertest1"
  val password = "acctpaymer*1"

  try {
    Class.forName(driverClass)
  } catch {
    case e: ClassNotFoundException => throw e
    case e: Exception => throw e
  }

  @throws(classOf[SQLException])
  def getConnection: Connection = {
    DriverManager.getConnection(jdbcUrl, username, password)
  }

  @throws(classOf[SQLException])
  def doTrancation(trancations: Set[String]): Unit = {
    val connection = getConnection
    connection.setAutoCommit(false)
    trancations.foreach(connection.createStatement().execute(_))
    connection.commit
    connection.close
  }
} 

//设置package包名称以及导入依赖的类
package pl.gigiel.vocabulary.tool

import java.sql.DriverManager
import java.sql.Connection
import org.h2.tools.Server

object SimpleApp extends App {
  //val fileName = "E:\\_tgl\\workspace-tomaszacer\\vocabulary-tool-confidential\\resources\\jack-london-white-fang.txt"
  //val all = words(fileName)
  //println(all.length)
  //println(all)

  db
  println("ok")

  def words(fileName: String): List[String] = {
    val source = scala.io.Source.fromFile(fileName)
    val wordPattern = """([A-Za-z\-'])+""".r
    val words = try source.getLines.flatMap(wordPattern.findAllIn).map(_.toLowerCase).toList.distinct.sorted finally source.close()

    return words
  }

  def db: Unit = {
    val driver = "org.h2.Driver"
    val url = "jdbc:h2:mem:vocabulary-db;DB_CLOSE_DELAY=-1;DATABASE_TO_UPPER=false"
    val username = "vocabulary-username"
    val password = "vocabulary-password"

    var server: Server = null
    var connection: Connection = null

    try {
      Class.forName(driver) // unnecessary in jdbc 4+
      server = Server.createTcpServer("-tcpPort", "9123", "-tcpAllowOthers").start
      // jdbc:h2:tcp://localhost:9123/mem:vocabulary-db;DB_CLOSE_DELAY=-1;DATABASE_TO_UPPER=false
      connection = DriverManager.getConnection(url, username, password)

      val stmt = connection.createStatement()
      stmt.execute("create table infos(id int primary key, info varchar(255))")
      stmt.execute("insert into infos(id, info) values(1, 'abc')")
      val rs = stmt.executeQuery("select id, info from infos")
      while (rs.next()) {
        val id = rs.getString("id")
        val info = rs.getString("info")
        println("id, info = " + id + ", " + info)
      }
      io.StdIn.readInt
    } catch {
      case e: Throwable => e.printStackTrace
    } finally {
      connection.close
      server.stop
    }
  }
} 

//设置package包名称以及导入依赖的类
package com.example.db


import java.sql.{Connection, DriverManager}

import com.twitter.util.Future
import org.postgresql.util.HostSpec


object DB {
  val url = "jdbc:postgresql://localhost:5432/postgres?user=Sergey&password=root"
  val connection = DriverManager.getConnection(url)

  def isValidToken(token: String) = {
    val statement = connection.prepareStatement("SELECT access_token FROM user_access WHERE access_token = ?")
    statement.setString(1, token)
    Future {
      statement.executeQuery()
    } map { resultSet =>
      resultSet.next()
    }
  }

  def createToken(userId: BigInt): Future[String] = {
    val token = java.util.UUID.randomUUID().toString
    val statement = connection.prepareStatement("INSERT INTO user_access VALUES (?, ?)")
    statement.setBigDecimal(2, BigDecimal(userId).bigDecimal)
    statement.setString(1, token)
    Future {
      statement.executeUpdate()
    } map {
      case 1 => token
      case _ => ""
    }
  }
} 

//设置package包名称以及导入依赖的类
package db.core

import java.sql.DriverManager
import db.config._
import java.sql.Connection
import snaq.db.ConnectionPool

case class DB(c:TraitDBConfig) {
  
  lazy val pool = new ConnectionPool("poolname", 1, 200, 300, 180000, c.url, c.dbuser, c.dbpass) // pool parameters hardcoded for now
  var connection:Connection = _
  def getConnection = if (c.usePool) pool.getConnection(3000) else {
    if (connection == null || !connection.isValid(3000)) {
      connection = DriverManager.getConnection(c.url, c.dbuser, c.dbpass)      
    }
    connection  
  }
} 

//设置package包名称以及导入依赖的类
package com.datawizards.dbtable2class.dialects

import java.sql.{DriverManager, ResultSet}
import org.apache.log4j.Logger
import com.datawizards.dbtable2class.model.ColumnMetadata
import scala.collection.mutable.ListBuffer

trait Dialect {
  protected val log: Logger = Logger.getLogger(getClass.getName)

  def mapColumnTypeToScalaType(column: ColumnMetadata): String

  def extractTableColumns(dbUrl: String, connectionProperties: java.util.Properties, database: String, schema: String, table: String): Iterable[ColumnMetadata] = {
    Class.forName(driverClassName)
    val connection = DriverManager.getConnection(dbUrl, connectionProperties)
    val query = extractTableColumnsQuery(database, schema, table)
    val rs = connection.createStatement().executeQuery(query)
    val buffer = new ListBuffer[ColumnMetadata]
    while(rs.next()) {
      buffer += extractColumnMetadata(rs)
    }
    connection.close()

    val extractedColumns = buffer.toList
    if(extractedColumns.size == 0){
      throw new RuntimeException(s"No columns found for specified table ${database}.${schema}.${table}. Please check if table is correct.")
    }
    extractedColumns
  }

  protected def extractColumnMetadata(rs: ResultSet): ColumnMetadata =
    ColumnMetadata(
      columnName = rs.getString(columnWithColumnName),
      typeName = rs.getString(columnWithTypeName)
    )

  protected def driverClassName: String
  protected def extractTableColumnsQuery(database: String, schema: String, table: String): String
  protected def columnWithColumnName: String
  protected def columnWithTypeName: String

} 

//设置package包名称以及导入依赖的类
package domain

import java.sql.DriverManager

import adapter.dto.ConnectDBDto

case class DBConnection(driver: String, url: String) {
  def getConnection = {
    Class.forName(driver).newInstance()
    DriverManager.getConnection(url)
  }
}

object DBConnection {
  def apply(dto: ConnectDBDto): DBConnection = dto.dbName match {
    case "MySQL" =>
      DBConnection(
        "com.mysql.jdbc.Driver",
        s"jdbc:mysql://${dto.host}:${dto.port}/?user=${dto.user}&password=${dto.pass.getOrElse("")}"
      )
    case _ => throw new Exception
  }

  val CACHE_KEY = "db.connection"

} 

//设置package包名称以及导入依赖的类
package com.aronia.core

import java.sql.{Connection, DriverManager}

import com.typesafe.scalalogging.LazyLogging
import com.aronia.core.AroniaUtils._
import com.aronia.core.spiders.GithubStargazers
import us.codecraft.webmagic.Spider

import scala.collection.mutable.ArrayBuffer


private[aronia] class AroniaContext extends LazyLogging {

  var spiders = new ArrayBuffer[Spider]
  val config = new AroniaConfig(true)
  private val defaultGlobalConfig = "ads_env.conf"

  // read environment variables from external file
  val gitConfig = {
    val configFile = config.get(config.ADS_ENV_CONFIG, null)
    if (configFile == null) {
      logger.info(s"Using default ads configurations ($defaultGlobalConfig)")
      loadKeyValueResource(defaultGlobalConfig)
    } else {
      logger.info(s"Using user-specified configuration file ($configFile)")
      loadKeyValueFile(configFile)
    }
  }

  config.setAll(gitConfig)
  // TODO: determine the priority of jvm and config file

  // global mysql backend
  classOf[com.mysql.jdbc.Driver]
  private val host = config.get(config.MYSQL_HOST, "127.0.0.1")
  private val port = config.get(config.MYSQL_PORT, "3306")
  private val username = config.get(config.MYSQL_USERNAME, "root")
  private val passwd = config.get(config.MYSQL_PASSWORD, "password")
  private val jdbcAddr = s"jdbc:mysql://$host:$port?user=$username&password=$passwd"

  logger.info(s"Connecting to $jdbcAddr")
  val mysqlBackend: Connection = DriverManager.getConnection(jdbcAddr)

  // configure all spiders
  private val githubStarSpider = (new GithubStargazers)
    .createSpider(this, Array(config.GITHUB_AWESOME_REPO), 10)
  spiders += githubStarSpider

  def go(): Unit = {
    logger.info("Running all spiders")
    spiders.foreach(_.run)
  }

  def clear(): Unit = {
    if (mysqlBackend != null)
      mysqlBackend.close()
  }

} 

//设置package包名称以及导入依赖的类
import java.sql.DriverManager
import java.sql.Connection

object JdbcSqlite {
  def main(args: Array[String]) {
    var c: Connection = null
    try {
      Class.forName("org.sqlite.JDBC")
      c = DriverManager.getConnection("jdbc:sqlite:planets.sqlite")
    } catch {
      case e: Throwable => e.printStackTrace
    }
    c.close()
  }
} 

//设置package包名称以及导入依赖的类
package io.sbelkin.jdbc.csvresolution


import java.sql.{Connection, DriverManager}
object Csv {

  var connection:Connection = null
  var path: String = null

  def main(path:String) : Unit = {
    setPath(path)
    select()
  }

  def setPath(path:String): Unit ={
    this.path = path
  }

  def connectionStart() : Unit = {
    if (path.length == 0) {
      throw new IllegalStateException("Must have a path for CSV Jdbc set.")
    } else {
      println("Using following path %s, as the data source.".format(path))
    }
    val driver:String = "org.relique.jdbc.csv.CsvDriver"
    Class.forName(driver)
    connection = DriverManager.getConnection("jdbc:relique:csv:" + path)
  }

  def connectionEnd() : Unit = {
    connection.close()
  }

  def select() : Unit = {
    connectionStart()
    try {
      val statement = connection.createStatement()
      val resultSet = statement.executeQuery("SELECT id,name FROM test")
      while ( resultSet.next() ) {
        val id = resultSet.getString("id")
        val name = resultSet.getString("name")
        println("id, name = " + id + ", " + name)
      }
    } catch {
      case e => e.printStackTrace
    } finally {
      connectionEnd()
    }
  }
} 

//设置package包名称以及导入依赖的类
package io.sbelkin.jdbc.csvresolution

import java.sql.{Connection, DriverManager}

import com.typesafe.config.Config


object Mysql {

  var connection:Connection = null
  var config:Config = null

  def main(config: Config): Unit = {
    setConfig(config);
    select()
  }

  def setConfig(config: Config) : Unit = {
    this.config = config;
  }

  def connectionStart() : Unit = {
    Class.forName(config.getString("database.driverClass"))
    val url = config.getString("database.url") +"/"+ config.getString("database.database")
    connection = DriverManager.getConnection(url, config.getString("database.user"), config.getString("database.password"))
  }
  def connectionClose() : Unit = {
    connection.close()
  }

  def select(): Unit =  {
    connectionStart()
    try {
      val statement = connection.createStatement()
      val resultSet = statement.executeQuery("SELECT host, user FROM user")
      while ( resultSet.next() ) {
        val host = resultSet.getString("host")
        val user = resultSet.getString("user")
        println("host, user = " + host + ", " + user)
      }
    } catch {
      case e => e.printStackTrace
    } finally {
      connectionClose()
    }
  }
} 

//设置package包名称以及导入依赖的类
package petclinic

import cats.data.EitherT
import com.typesafe.config.ConfigFactory
import scala.concurrent.{ ExecutionContext, Future }
import scala.util.{ Failure, Success, Try }
import scala.util.control.NonFatal
import scala.sys.process._
import java.sql.{ Connection, DriverManager }

package object mysql {

  final val Config = ConfigFactory.load()

  final val Ip =
    Try("docker-machine ip".!!).orElse(Try(Config.getString("mysql.ip"))).getOrElse("localhost")
  final val Driver   = "com.mysql.jdbc.Driver"
  final val Url      = s"jdbc:mysql://$Ip/petclinic"
  final val username = Try(Config.getString("mysql.username")).getOrElse("root")
  final val password = Try(Config.getString("mysql.password")).getOrElse("root")

  def withConnection[A](f: Connection => Either[PetClinicError, A])(
      implicit ec: ExecutionContext): Response[A] =
    EitherT(Future {
      var connection: Connection = null
      val comp =
        try {
          connection = DriverManager.getConnection(Url, username, password)
          Success(f(connection))
        } catch {
          case NonFatal(e) => Failure(e)
        } finally {
          if (connection != null && !connection.isClosed)
            connection.close()
        }
      comp.get
    })
} 

//设置package包名称以及导入依赖的类
package io.eels.component.jdbc

import java.sql.{Connection, DriverManager, ResultSet}

import com.sksamuel.exts.Logging
import io.eels.component.jdbc.dialect.JdbcDialect
import io.eels.schema.StructType

trait JdbcPrimitives extends Logging {

  def connect(url: String): Connection = {
    logger.debug(s"Connecting to jdbc source $url...")
    val conn = DriverManager.getConnection(url)
    logger.debug(s"Connected to $url")
    conn
  }

  def schemaFor(dialect: JdbcDialect, rs: ResultSet): StructType = {
    val schema = JdbcSchemaFns.fromJdbcResultset(rs, dialect)
    logger.trace("Fetched schema:\n" + schema.show())
    schema
  }
} 

//设置package包名称以及导入依赖的类
package io.eels.component.jdbc

import java.sql.{Connection, DriverManager}

import com.sksamuel.exts.Logging
import com.typesafe.config.ConfigFactory
import io.eels.Sink
import io.eels.component.jdbc.dialect.{GenericJdbcDialect, JdbcDialect}
import io.eels.schema.StructType
import com.sksamuel.exts.OptionImplicits._

object JdbcSink extends Logging {

  private val config = ConfigFactory.load()
  private val warnIfMissingRewriteBatchedStatements = config.getBoolean("eel.jdbc.sink.warnIfMissingRewriteBatchedStatements")

  def apply(url: String, table: String): JdbcSink = {
    if (!url.contains("rewriteBatchedStatements")) {
      if (warnIfMissingRewriteBatchedStatements) {
        logger.warn("JDBC connection string does not contain the property 'rewriteBatchedStatements=true' which can be a major performance boost when writing data via JDBC. " +
          "Add this property to your connection string, or to remove this warning set eel.jdbc.warnIfMissingRewriteBatchedStatements=false")
      }
    }
    JdbcSink(() => DriverManager.getConnection(url), table)
  }
}

case class JdbcSink(connFn: () => Connection,
                    table: String,
                    createTable: Boolean = false,
                    batchSize: Int = 1000, // the number of rows before a commit is made
                    batchesPerCommit: Int = 0, // 0 means commit at the end, otherwise how many batches before a commit
                    dialect: Option[JdbcDialect] = None,
                    threads: Int = 4) extends Sink with Logging {

  private val config = ConfigFactory.load()
  private val bufferSize = config.getInt("eel.jdbc.sink.bufferSize")
  private val autoCommit = config.getBoolean("eel.jdbc.sink.autoCommit")

  def withCreateTable(createTable: Boolean): JdbcSink = copy(createTable = createTable)
  def withBatchSize(batchSize: Int): JdbcSink = copy(batchSize = batchSize)
  def withThreads(threads: Int): JdbcSink = copy(threads = threads)
  def withBatchesPerCommit(commitSize: Int): JdbcSink = copy(batchesPerCommit = batchesPerCommit)
  def withDialect(dialect: JdbcDialect): JdbcSink = copy(dialect = dialect.some)

  override def open(schema: StructType) =
    new JdbcSinkWriter(schema, connFn, table, createTable, dialect.getOrElse(new GenericJdbcDialect), threads, batchSize, batchesPerCommit, autoCommit, bufferSize)
} 

//设置package包名称以及导入依赖的类
package io.eels.component.jdbc

import java.sql.{Connection, DriverManager, PreparedStatement}

import com.sksamuel.exts.Logging
import com.sksamuel.exts.io.Using
import com.sksamuel.exts.metrics.Timed
import io.eels.{Row, Source}
import io.eels.component.jdbc.dialect.{GenericJdbcDialect, JdbcDialect}
import io.eels.datastream.Publisher
import io.eels.schema.StructType

object JdbcSource {
  def apply(url: String, query: String): JdbcSource = JdbcSource(() => DriverManager.getConnection(url), query)
}

case class JdbcSource(connFn: () => Connection,
                      query: String,
                      bindFn: (PreparedStatement) => Unit = stmt => (),
                      fetchSize: Int = 200,
                      providedSchema: Option[StructType] = None,
                      providedDialect: Option[JdbcDialect] = None,
                      partitionStrategy: JdbcPartitionStrategy = SinglePartitionStrategy)
  extends Source with JdbcPrimitives with Logging with Using with Timed {

  override lazy val schema: StructType = providedSchema.getOrElse(fetchSchema())

  def withBind(bind: (PreparedStatement) => Unit): JdbcSource = copy(bindFn = bind)
  def withFetchSize(fetchSize: Int): JdbcSource = copy(fetchSize = fetchSize)
  def withProvidedSchema(schema: StructType): JdbcSource = copy(providedSchema = Option(schema))
  def withProvidedDialect(dialect: JdbcDialect): JdbcSource = copy(providedDialect = Option(dialect))
  def withPartitionStrategy(strategy: JdbcPartitionStrategy): JdbcSource = copy(partitionStrategy = strategy)

  private def dialect(): JdbcDialect = providedDialect.getOrElse(new GenericJdbcDialect())

  override def parts(): Seq[Publisher[Seq[Row]]] = partitionStrategy.parts(connFn, query, bindFn, fetchSize, dialect())

  def fetchSchema(): StructType = {
    using(connFn()) { conn =>
      val schemaQuery = s"SELECT * FROM ($query) tmp WHERE 1=0"
      using(conn.prepareStatement(schemaQuery)) { stmt =>

        stmt.setFetchSize(fetchSize)
        bindFn(stmt)

        val rs = timed(s"Executing query $query") {
          stmt.executeQuery()
        }

        val schema = schemaFor(dialect(), rs)
        rs.close()
        schema
      }
    }
  }
} 

//设置package包名称以及导入依赖的类
package io.eels.component.jdbc

import java.sql.DriverManager

import io.eels.schema._
import org.scalatest.{FlatSpec, Matchers}

// tests that the schema mappings for jdbc are correct using h2
class H2SchemaCompatibilityTest extends FlatSpec with Matchers {

  Class.forName("org.h2.Driver")

  val conn = DriverManager.getConnection("jdbc:h2:mem:schematest")
  conn.createStatement().executeUpdate("create table t (" +
    "a integer, " +
    "b bit, " +
    "c bigint, " +
    "d double, " +
    "e time, " +
    "f date, " +
    "g timestamp, " +
    "h decimal(14,5)," +
    "i text," +
    "j smallint," +
    "k tinyint," +
    "l varchar(55)," +
    "m char(15)," +
    "n numeric(66,5)" +
    ")")

  "JdbcSource" should "map schemas from jdbc to correct eel types" in {
    JdbcSource(() => conn, "select * from t").schema shouldBe
      StructType(
        Vector(
          Field("A", IntType(true), true, false),
          Field("B", io.eels.schema.BooleanType, true, false),
          Field("C", io.eels.schema.BigIntType, true, false),
          Field("D", io.eels.schema.DoubleType, true, false),
          Field("E", io.eels.schema.TimeMillisType, true, false),
          Field("F", io.eels.schema.DateType, true, false),
          Field("G", io.eels.schema.TimestampMillisType, true, false),
          Field("H", DecimalType(Precision(14), Scale(5)), true, false),
          Field("I", io.eels.schema.StringType, true, false),
          Field("J", io.eels.schema.ShortType(true), true, false),
          Field("K", io.eels.schema.ShortType(true), true, false),
          Field("L", io.eels.schema.VarcharType(55), true, false),
          Field("M", io.eels.schema.CharType(15), true, false),
          Field("N", DecimalType(Precision(66), Scale(5)), true, false)
        )
      )
  }
} 

//设置package包名称以及导入依赖的类
package io.eels.component.jdbc

import java.sql.DriverManager

import org.scalatest.{Matchers, WordSpec}

import scala.util.Random

class RangePartitionStrategyTest extends WordSpec with Matchers {

  Class.forName("org.h2.Driver")

  private val conn = DriverManager.getConnection("jdbc:h2:mem:rangetest")
  conn.createStatement().executeUpdate("create table bucket_test (a integer)")
  for (k <- 0 until 20) {
    conn.createStatement().executeUpdate(s"insert into bucket_test (a) values (${Random.nextInt(10000)})")
  }

  "BucketPartitionStrategy" should {
    "generate evenly spaced ranges" in {
      RangePartitionStrategy("a", 10, 2, 29).ranges shouldBe List((2, 4), (5, 7), (8, 10), (11, 13), (14, 16), (17, 19), (20, 22), (23, 25), (26, 27), (28, 29))
      RangePartitionStrategy("a", 2, 2, 30).ranges shouldBe List((2, 16), (17, 30))
      RangePartitionStrategy("a", 1, 4, 5).ranges shouldBe List((4, 5))
      RangePartitionStrategy("a", 1, 4, 4).ranges shouldBe List((4, 4))
      RangePartitionStrategy("a", 6, 1, 29).ranges shouldBe List((1, 5), (6, 10), (11, 15), (16, 20), (21, 25), (26, 29))
    }
    "return correct number of ranges" in {
      JdbcSource(() => DriverManager.getConnection("jdbc:h2:mem:rangetest"), "select * from bucket_test")
        .withPartitionStrategy(RangePartitionStrategy("a", 4, 0, 10000))
        .parts().size shouldBe 4
    }
    "return full and non overlapping data" in {
      JdbcSource(() => DriverManager.getConnection("jdbc:h2:mem:rangetest"), "select * from bucket_test")
        .withPartitionStrategy(RangePartitionStrategy("a", 4, 0, 10000))
        .toDataStream().collect.size shouldBe 20
    }
  }
} 

//设置package包名称以及导入依赖的类
package io.eels.component.jdbc

import java.sql.DriverManager
import java.util.UUID

import io.eels.Row
import io.eels.datastream.DataStream
import io.eels.schema.{Field, StructType}
import org.scalatest.{Matchers, OneInstancePerTest, WordSpec}

class JdbcSinkTest extends WordSpec with Matchers with OneInstancePerTest {

  Class.forName("org.h2.Driver")
  private val url = "jdbc:h2:mem:" + UUID.randomUUID.toString.replace("-", "")
  private val conn = DriverManager.getConnection(url)
  conn.createStatement().executeUpdate("create table mytab (a integer, b integer, c integer)")

  private val schema = StructType(Field("a"), Field("b"), Field("c"))
  private val frame = DataStream.fromRows(schema, Row(schema, Vector("1", "2", "3")), Row(schema, Vector("4", "5", "6")))

  "JdbcSink" should {
    "write frame to table" in {
      frame.to(JdbcSink(url, "mytab"))
      val rs = conn.createStatement().executeQuery("select count(*) from mytab")
      rs.next()
      rs.getLong(1) shouldBe 2L
      rs.close()
    }
    "create table if createTable is true" in {
      frame.to(JdbcSink(url, "qwerty").withCreateTable(true))
      val rs = conn.createStatement().executeQuery("select count(*) from qwerty")
      rs.next()
      rs.getLong(1) shouldBe 2L
      rs.close()
    }
    "support multiple writers" in {
      val rows = List.fill(10000)(Row(schema, Vector("1", "2", "3")))
      val mframe = DataStream.fromRows(schema, rows)
      val sink = JdbcSink(url, "multithreads").withCreateTable(true).withThreads(4)
      mframe.to(sink)
      val rs = conn.createStatement().executeQuery("select count(*) from multithreads")
      rs.next()
      rs.getLong(1) shouldBe 10000L
      rs.close()
    }
  }
}