/*
 * 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.spark.carbondata

import java.io.{File, PrintWriter}
import java.math.BigDecimal
import java.net.{BindException, ServerSocket}
import java.sql.{Date, Timestamp}

import scala.collection.mutable

import org.apache.spark.sql._
import org.apache.spark.sql.catalyst.TableIdentifier
import org.apache.spark.sql.hive.CarbonRelation
import org.apache.spark.sql.streaming.{ProcessingTime, StreamingQuery}
import org.apache.spark.sql.test.util.QueryTest
import org.scalatest.BeforeAndAfterAll

import org.apache.carbondata.core.constants.CarbonCommonConstants
import org.apache.carbondata.core.metadata.schema.table.CarbonTable
import org.apache.carbondata.core.util.CarbonProperties
import org.apache.carbondata.core.util.path.CarbonTablePath
import org.apache.carbondata.streaming.parser.CarbonStreamParser

case class StreamLongStrData(id: Integer, name: String, city: String, salary: java.lang.Float,
    tax: BigDecimal, percent: java.lang.Double, birthday: String,
    register: String, updated: String, longStr: String,
    file: FileElement)

class TestStreamingTableWithLongString extends QueryTest with BeforeAndAfterAll {

  private val spark = sqlContext.sparkSession
  private val dataFilePath = s"$resourcesPath/streamSample_with_long_string.csv"
  private val csvDataDir = integrationPath + "/spark/target/csvdata_longstr"
  private val longStrValue = "abc" * 12000
  // scalastyle:off lineLength
  override def beforeAll {
    CarbonProperties.getInstance().addProperty(
      CarbonCommonConstants.CARBON_TIMESTAMP_FORMAT,
      CarbonCommonConstants.CARBON_TIMESTAMP_DEFAULT_FORMAT)
    CarbonProperties.getInstance().addProperty(
      CarbonCommonConstants.CARBON_DATE_FORMAT,
      CarbonCommonConstants.CARBON_DATE_DEFAULT_FORMAT)
    sql("DROP DATABASE IF EXISTS streaming_longstr CASCADE")
    sql("CREATE DATABASE streaming_longstr")
    sql("USE streaming_longstr")

    dropTable()

    // 1. streaming table with long string field
    // socket source
    createTable(tableName = "stream_table_longstr", streaming = true, withBatchLoad = true)

    // 2. streaming table with long string field
    // file source
    createTable(tableName = "stream_table_longstr_file", streaming = true, withBatchLoad = true)

    // 3. streaming table with long string and complex field
    createTableWithComplexType(
      tableName = "stream_table_longstr_complex", streaming = true, withBatchLoad = true)
  }

  override def afterAll {
    dropTable()
    sql("USE default")
    sql("DROP DATABASE IF EXISTS streaming_longstr CASCADE")
    new File(csvDataDir).delete()
  }

  def dropTable(): Unit = {
    sql("drop table if exists streaming_longstr.stream_table_longstr")
    sql("drop table if exists streaming_longstr.stream_table_longstr_file")
    sql("drop table if exists streaming_longstr.stream_table_longstr_complex")
  }

  // input source: file
  test("[CARBONDATA-3497] Support to write long string for streaming table: " +
       "ingest from file source") {
    val identifier = new TableIdentifier("stream_table_longstr_file", Option("streaming_longstr"))
    val carbonTable = CarbonEnv.getInstance(spark).carbonMetaStore.lookupRelation(identifier)(spark)
      .asInstanceOf[CarbonRelation].carbonTable
    // streaming ingest 10 rows
    generateCSVDataFile(spark, idStart = 10, rowNums = 10, csvDataDir)
    val thread = createFileStreamingThread(spark, carbonTable, csvDataDir, intervalSecond = 1,
      identifier)
    thread.start()
    Thread.sleep(3000)
    generateCSVDataFile(spark, idStart = 30, rowNums = 10, csvDataDir)
    Thread.sleep(5000)
    thread.interrupt()
    checkAnswer(
      sql("select count(*) from streaming_longstr.stream_table_longstr_file"),
      Seq(Row(25))
    )

    val row = sql("select * from streaming_longstr.stream_table_longstr_file order by id").head()
    val exceptedRow = Row(10, "name_10", "city_10", 100000.0, BigDecimal.valueOf(0.01), 80.01, Date.valueOf("1990-01-01"), Timestamp.valueOf("2010-01-01 10:01:01.0"), Timestamp.valueOf("2010-01-01 10:01:01.0"), "10" + longStrValue)
    assertResult(exceptedRow)(row)
    new File(csvDataDir).delete()
  }

  test("[CARBONDATA-3497] Support to write long string for streaming table") {
    executeStreamingIngest(
      tableName = "stream_table_longstr",
      batchNums = 2,
      rowNumsEachBatch = 25,
      intervalOfSource = 5,
      intervalOfIngest = 5,
      continueSeconds = 20,
      handoffSize = 51200,
      autoHandoff = false
    )

    var result = sql("select * from streaming_longstr.stream_table_longstr order by id, name")
      .collect()
    assert(result != null)
    assert(result.length == 55)
    // check one row of streaming data
    assert(result(1).getString(1) == "name_2")
    assert(result(1).getString(9) == ("2" + longStrValue))
    // check one row of batch loading
    assert(result(50).getInt(0) == 100000001)
    assert(result(50).getString(1) == "batch_1")
    assert(result(50).getString(9) == ("1" + longStrValue))

    checkAnswer(
      sql("select * from streaming_longstr.stream_table_longstr where id = 1"),
      Seq(Row(1, "name_1", "city_1", 10000.0, BigDecimal.valueOf(0.01), 80.01, Date.valueOf("1990-01-01"), Timestamp.valueOf("2010-01-01 10:01:01.0"), Timestamp.valueOf("2010-01-01 10:01:01.0"), ("1" + longStrValue))))

    checkAnswer(
      sql("select * from streaming_longstr.stream_table_longstr where id > 49 and id < 100000002"),
      Seq(Row(50, "name_50", "city_50", 500000.0, BigDecimal.valueOf(0.01), 80.01, Date.valueOf("1990-01-01"), Timestamp.valueOf("2010-01-01 10:01:01.0"), Timestamp.valueOf("2010-01-01 10:01:01.0"), ("50" + longStrValue)),
        Row(100000001, "batch_1", "city_1", 0.1, BigDecimal.valueOf(0.01), 80.01, Date.valueOf("1990-01-01"), Timestamp.valueOf("2010-01-01 10:01:01.0"), Timestamp.valueOf("2010-01-01 10:01:01.0"), ("1" + longStrValue))))

    checkAnswer(
      sql("select * from streaming_longstr.stream_table_longstr where id between 50 and 100000001"),
      Seq(Row(50, "name_50", "city_50", 500000.0, BigDecimal.valueOf(0.01), 80.01, Date.valueOf("1990-01-01"), Timestamp.valueOf("2010-01-01 10:01:01.0"), Timestamp.valueOf("2010-01-01 10:01:01.0"), ("50" + longStrValue)),
        Row(100000001, "batch_1", "city_1", 0.1, BigDecimal.valueOf(0.01), 80.01, Date.valueOf("1990-01-01"), Timestamp.valueOf("2010-01-01 10:01:01.0"), Timestamp.valueOf("2010-01-01 10:01:01.0"), ("1" + longStrValue))))

    sql("show segments for table streaming_longstr.stream_table_longstr").collect()
    sql("alter table streaming_longstr.stream_table_longstr finish streaming")
    sql("alter table streaming_longstr.stream_table_longstr compact 'streaming'")
    sql("show segments for table streaming_longstr.stream_table_longstr").collect()
    Thread.sleep(5000)

    result = sql("select * from streaming_longstr.stream_table_longstr order by id, name").collect()
    assert(result != null)
    assert(result.length == 55)
    // check one row of streaming data
    assert(result(2).getString(1) == "name_3")
    assert(result(2).getString(9) == ("3" + longStrValue))
    // check one row of batch loading
    assert(result(51).getInt(0) == 100000002)
    assert(result(51).getString(1) == "batch_2")
    assert(result(51).getString(9) == ("2" + longStrValue))

    checkAnswer(
      sql("select * from streaming_longstr.stream_table_longstr where id = 1"),
      Seq(Row(1, "name_1", "city_1", 10000.0, BigDecimal.valueOf(0.01), 80.01, Date.valueOf("1990-01-01"), Timestamp.valueOf("2010-01-01 10:01:01.0"), Timestamp.valueOf("2010-01-01 10:01:01.0"), ("1" + longStrValue))))

    checkAnswer(
      sql("select * from streaming_longstr.stream_table_longstr where id > 49 and id < 100000002"),
      Seq(Row(50, "name_50", "city_50", 500000.0, BigDecimal.valueOf(0.01), 80.01, Date.valueOf("1990-01-01"), Timestamp.valueOf("2010-01-01 10:01:01.0"), Timestamp.valueOf("2010-01-01 10:01:01.0"), ("50" + longStrValue)),
        Row(100000001, "batch_1", "city_1", 0.1, BigDecimal.valueOf(0.01), 80.01, Date.valueOf("1990-01-01"), Timestamp.valueOf("2010-01-01 10:01:01.0"), Timestamp.valueOf("2010-01-01 10:01:01.0"), ("1" + longStrValue))))

    checkAnswer(
      sql("select * from streaming_longstr.stream_table_longstr where id between 50 and 100000001"),
      Seq(Row(50, "name_50", "city_50", 500000.0, BigDecimal.valueOf(0.01), 80.01, Date.valueOf("1990-01-01"), Timestamp.valueOf("2010-01-01 10:01:01.0"), Timestamp.valueOf("2010-01-01 10:01:01.0"), ("50" + longStrValue)),
        Row(100000001, "batch_1", "city_1", 0.1, BigDecimal.valueOf(0.01), 80.01, Date.valueOf("1990-01-01"), Timestamp.valueOf("2010-01-01 10:01:01.0"), Timestamp.valueOf("2010-01-01 10:01:01.0"), ("1" + longStrValue))))

    sql("alter table streaming_longstr.stream_table_longstr compact 'major'")
    sql("show segments for table streaming_longstr.stream_table_longstr").collect()
    Thread.sleep(5000)

    result = sql("select * from streaming_longstr.stream_table_longstr order by id, name").collect()
    assert(result != null)
    assert(result.length == 55)
    // check one row of streaming data
    assert(result(3).getString(1) == "name_4")
    assert(result(3).getString(9) == ("4" + longStrValue))
    // check one row of batch loading
    assert(result(52).getInt(0) == 100000003)
    assert(result(52).getString(1) == "batch_3")
    assert(result(52).getString(9) == ("3" + longStrValue))

    checkAnswer(
      sql("select * from streaming_longstr.stream_table_longstr where id = 1"),
      Seq(Row(1, "name_1", "city_1", 10000.0, BigDecimal.valueOf(0.01), 80.01, Date.valueOf("1990-01-01"), Timestamp.valueOf("2010-01-01 10:01:01.0"), Timestamp.valueOf("2010-01-01 10:01:01.0"), ("1" + longStrValue))))

    checkAnswer(
      sql("select * from streaming_longstr.stream_table_longstr where id > 49 and id < 100000002"),
      Seq(Row(50, "name_50", "city_50", 500000.0, BigDecimal.valueOf(0.01), 80.01, Date.valueOf("1990-01-01"), Timestamp.valueOf("2010-01-01 10:01:01.0"), Timestamp.valueOf("2010-01-01 10:01:01.0"), ("50" + longStrValue)),
        Row(100000001, "batch_1", "city_1", 0.1, BigDecimal.valueOf(0.01), 80.01, Date.valueOf("1990-01-01"), Timestamp.valueOf("2010-01-01 10:01:01.0"), Timestamp.valueOf("2010-01-01 10:01:01.0"), ("1" + longStrValue))))

    checkAnswer(
      sql("select * from streaming_longstr.stream_table_longstr where id between 50 and 100000001"),
      Seq(Row(50, "name_50", "city_50", 500000.0, BigDecimal.valueOf(0.01), 80.01, Date.valueOf("1990-01-01"), Timestamp.valueOf("2010-01-01 10:01:01.0"), Timestamp.valueOf("2010-01-01 10:01:01.0"), ("50" + longStrValue)),
        Row(100000001, "batch_1", "city_1", 0.1, BigDecimal.valueOf(0.01), 80.01, Date.valueOf("1990-01-01"), Timestamp.valueOf("2010-01-01 10:01:01.0"), Timestamp.valueOf("2010-01-01 10:01:01.0"), ("1" + longStrValue))))
  }

  test("[CARBONDATA-3497] Support to write long string for streaming table: " +
       "include complex column") {
    executeStreamingIngest(
      tableName = "stream_table_longstr_complex",
      batchNums = 2,
      rowNumsEachBatch = 25,
      intervalOfSource = 5,
      intervalOfIngest = 5,
      continueSeconds = 20,
      handoffSize = 51200,
      autoHandoff = false
    )

    // non-filter
    val result = sql(
      "select * from streaming_longstr.stream_table_longstr_complex order by id, name").collect()
    assert(result != null)
    assert(result.length == 55)
    // check one row of streaming data
    assert(result(3).getString(1) == "name_4")
    assert(result(3).getString(9) == ("4" + longStrValue))
    // check one row of batch loading
    assert(result(52).getInt(0) == 100000003)
    assert(result(52).getString(1) == "batch_3")
    assert(result(52).getString(9) == ("3" + longStrValue))
    assert(result(52).getStruct(10).getInt(1) == 40)

    // filter
    checkAnswer(
      sql("select * from streaming_longstr.stream_table_longstr_complex where id = 1"),
      Seq(Row(1, "name_1", "city_1", 10000.0, BigDecimal.valueOf(0.01), 80.01, Date.valueOf("1990-01-01"), Timestamp.valueOf("2010-01-01 10:01:01.0"), Timestamp.valueOf("2010-01-01 10:01:01.0"), ("1" + longStrValue), Row(wrap(Array("school_1", "school_11")), 1))))

    checkAnswer(
      sql("select * from streaming_longstr.stream_table_longstr_complex where id > 49 and id < 100000002"),
      Seq(Row(50, "name_50", "city_50", 500000.0, BigDecimal.valueOf(0.01), 80.01, Date.valueOf("1990-01-01"), Timestamp.valueOf("2010-01-01 10:01:01.0"), Timestamp.valueOf("2010-01-01 10:01:01.0"), ("50" + longStrValue), Row(wrap(Array("school_50", "school_5050")), 50)),
        Row(100000001, "batch_1", "city_1", 0.1, BigDecimal.valueOf(0.01), 80.01, Date.valueOf("1990-01-01"), Timestamp.valueOf("2010-01-01 10:01:01.0"), Timestamp.valueOf("2010-01-01 10:01:01.0"), ("1" + longStrValue), Row(wrap(Array("school_1", "school_11")), 20))))

    checkAnswer(
      sql("select * from streaming_longstr.stream_table_longstr_complex where id between 50 and 100000001"),
      Seq(Row(50, "name_50", "city_50", 500000.0, BigDecimal.valueOf(0.01), 80.01, Date.valueOf("1990-01-01"), Timestamp.valueOf("2010-01-01 10:01:01.0"), Timestamp.valueOf("2010-01-01 10:01:01.0"), ("50" + longStrValue), Row(wrap(Array("school_50", "school_5050")), 50)),
        Row(100000001, "batch_1", "city_1", 0.1, BigDecimal.valueOf(0.01), 80.01, Date.valueOf("1990-01-01"), Timestamp.valueOf("2010-01-01 10:01:01.0"), Timestamp.valueOf("2010-01-01 10:01:01.0"), ("1" + longStrValue), Row(wrap(Array("school_1", "school_11")), 20))))
  }

  test("[CARBONDATA-3497] Support to write long string for streaming table: StreamSQL") {
    sql("DROP TABLE IF EXISTS source")
    sql("DROP TABLE IF EXISTS sink")

    var rows = sql("SHOW STREAMS").collect()
    assertResult(0)(rows.length)

    val csvDataDir = integrationPath + "/spark/target/streamSql_longstr"
    // streaming ingest 10 rows
    generateCSVDataFile(spark, idStart = 10, rowNums = 10, csvDataDir)

    sql(
      s"""
         |CREATE TABLE source(
         | id INT,
         | name STRING,
         | city STRING,
         | salary FLOAT,
         | tax DECIMAL(8,2),
         | percent double,
         | birthday DATE,
         | register TIMESTAMP,
         | updated TIMESTAMP,
         | longstr STRING
         |)
         |STORED AS carbondata
         |TBLPROPERTIES (
         | 'streaming'='source',
         | 'format'='csv',
         | 'path'='$csvDataDir'
         |)
      """.stripMargin)

    sql(
      s"""
         |CREATE TABLE sink(
         | id INT,
         | name STRING,
         | city STRING,
         | salary FLOAT,
         | tax DECIMAL(8,2),
         | percent double,
         | birthday DATE,
         | register TIMESTAMP,
         | updated TIMESTAMP,
         | longstr STRING
         | )
         |STORED AS carbondata
         |TBLPROPERTIES('streaming'='sink', 'LONG_STRING_COLUMNS'='longstr')
      """.stripMargin)

    sql(
      """
        |CREATE STREAM stream123 ON TABLE sink
        |STMPROPERTIES(
        |  'trigger'='ProcessingTime',
        |  'interval'='5 seconds')
        |AS
        |  SELECT *
        |  FROM source
        |  WHERE id % 2 = 1
      """.stripMargin).collect()

    Thread.sleep(200)
    sql("select * from sink").collect()

    generateCSVDataFile(spark, idStart = 30, rowNums = 10, csvDataDir, SaveMode.Append)
    Thread.sleep(7000)

    // after 2 minibatch, there should be 10 row added (filter condition: id%2=1)
    checkAnswer(sql("select count(*) from sink"), Seq(Row(10)))

    val row = sql("select * from sink order by id").head()
    val exceptedRow = Row(11, "name_11", "city_11", 110000.0, BigDecimal.valueOf(0.01), 80.01, Date.valueOf("1990-01-01"), Timestamp.valueOf("2010-01-01 10:01:01.0"), Timestamp.valueOf("2010-01-01 10:01:01.0"), ("11" + longStrValue))
    assertResult(exceptedRow)(row)

    sql("SHOW STREAMS").collect()

    rows = sql("SHOW STREAMS").collect()
    assertResult(1)(rows.length)
    assertResult("stream123")(rows.head.getString(0))
    assertResult("RUNNING")(rows.head.getString(2))
    assertResult("streaming_longstr.source")(rows.head.getString(3))
    assertResult("streaming_longstr.sink")(rows.head.getString(4))

    rows = sql("SHOW STREAMS ON TABLE sink").collect()
    assertResult(1)(rows.length)
    assertResult("stream123")(rows.head.getString(0))
    assertResult("RUNNING")(rows.head.getString(2))
    assertResult("streaming_longstr.source")(rows.head.getString(3))
    assertResult("streaming_longstr.sink")(rows.head.getString(4))

    sql("DROP STREAM stream123")
    sql("DROP STREAM IF EXISTS stream123")

    rows = sql("SHOW STREAMS").collect()
    assertResult(0)(rows.length)

    sql("DROP TABLE IF EXISTS source")
    sql("DROP TABLE IF EXISTS sink")
    new File(csvDataDir).delete()
  }

  def createWriteSocketThread(
      serverSocket: ServerSocket,
      writeNums: Int,
      rowNums: Int,
      intervalSecond: Int): Thread = {
    new Thread() {
      override def run(): Unit = {
        // wait for client to connection request and accept
        val clientSocket = serverSocket.accept()
        val socketWriter = new PrintWriter(clientSocket.getOutputStream())
        var index = 0
        for (_ <- 1 to writeNums) {
          // write 5 records per iteration
          val stringBuilder = new StringBuilder()
          for (_ <- 1 to rowNums) {
            index = index + 1
            stringBuilder.append(
              index.toString + ",name_" + index + ",city_" + index + "," +
              (10000.00 * index).toString + ",0.01,80.01" +
              ",1990-01-01,2010-01-01 10:01:01,2010-01-01 10:01:01," + index.toString() +
              ("abc" * 12000) + ",school_" + index + ":school_" + index + index + "$" + index)
            stringBuilder.append("\n")
          }
          socketWriter.append(stringBuilder.toString())
          socketWriter.flush()
          Thread.sleep(1000 * intervalSecond)
        }
        socketWriter.close()
      }
    }
  }

  def createSocketStreamingThread(
      spark: SparkSession,
      port: Int,
      carbonTable: CarbonTable,
      tableIdentifier: TableIdentifier,
      intervalSecond: Int = 2,
      handoffSize: Long = CarbonCommonConstants.HANDOFF_SIZE_DEFAULT,
      autoHandoff: Boolean = CarbonCommonConstants.ENABLE_AUTO_HANDOFF_DEFAULT.toBoolean
  ): Thread = {
    new Thread() {
      override def run(): Unit = {
        var qry: StreamingQuery = null
        try {
          import spark.implicits._
          val readSocketDF = spark.readStream
            .format("socket")
            .option("host", "localhost")
            .option("port", port)
            .load().as[String]
            .map(_.split(","))
            .map { fields => {
              val tmp = fields(10).split("\\$")
              val file = FileElement(tmp(0).split(":"), tmp(1).toInt)
              StreamLongStrData(fields(0).toInt, fields(1), fields(2), fields(3).toFloat,
                  BigDecimal.valueOf(fields(4).toDouble), fields(5).toDouble,
                  fields(6), fields(7), fields(8), fields(9), file)
            } }

          // Write data from socket stream to carbondata file
          // repartition to simulate an empty partition when readSocketDF has only one row
          qry = readSocketDF.repartition(2).writeStream
            .format("carbondata")
            .trigger(ProcessingTime(s"$intervalSecond seconds"))
            .option("checkpointLocation",
              CarbonTablePath.getStreamingCheckpointDir(carbonTable.getTablePath))
            .option("dbName", tableIdentifier.database.get)
            .option("tableName", tableIdentifier.table)
            .option(CarbonCommonConstants.HANDOFF_SIZE, handoffSize)
            .option("timestampformat", CarbonCommonConstants.CARBON_TIMESTAMP_DEFAULT_FORMAT)
            .option(CarbonCommonConstants.ENABLE_AUTO_HANDOFF, autoHandoff)
            .start()
          qry.awaitTermination()
        } catch {
          case ex: Throwable =>
            LOGGER.error(ex.getMessage)
            throw new Exception(ex.getMessage, ex)
        } finally {
          if (null != qry) {
            qry.stop()
          }
        }
      }
    }
  }

  /**
   * start ingestion thread: write `rowNumsEachBatch` rows repeatly for `batchNums` times.
   */
  def executeStreamingIngest(
      tableName: String,
      batchNums: Int,
      rowNumsEachBatch: Int,
      intervalOfSource: Int,
      intervalOfIngest: Int,
      continueSeconds: Int,
      handoffSize: Long = CarbonCommonConstants.HANDOFF_SIZE_DEFAULT,
      autoHandoff: Boolean = CarbonCommonConstants.ENABLE_AUTO_HANDOFF_DEFAULT.toBoolean
  ): Unit = {
    val identifier = new TableIdentifier(tableName, Option("streaming_longstr"))
    val carbonTable = CarbonEnv.getInstance(spark).carbonMetaStore.lookupRelation(identifier)(spark)
      .asInstanceOf[CarbonRelation].carbonTable
    var server: ServerSocket = null
    try {
      server = getServerSocket()
      val thread1 = createWriteSocketThread(
        serverSocket = server,
        writeNums = batchNums,
        rowNums = rowNumsEachBatch,
        intervalSecond = intervalOfSource)
      val thread2 = createSocketStreamingThread(
        spark = spark,
        port = server.getLocalPort,
        carbonTable = carbonTable,
        tableIdentifier = identifier,
        intervalSecond = intervalOfIngest,
        handoffSize = handoffSize,
        autoHandoff = autoHandoff)
      thread1.start()
      thread2.start()
      Thread.sleep(continueSeconds * 1000)
      thread2.interrupt()
      thread1.interrupt()
    } catch {
      case ex =>
        LOGGER.error("finished to ingest data", ex)
    } finally {
      if (null != server) {
        server.close()
      }
    }
  }

  def generateCSVDataFile(
      spark: SparkSession,
      idStart: Int,
      rowNums: Int,
      csvDirPath: String,
      saveMode: SaveMode = SaveMode.Overwrite): Unit = {
    // Create csv data frame file
    val csvDataDF = {
      // generate data with dimension columns (name and city)
      val csvRDD = spark.sparkContext.parallelize(idStart until idStart + rowNums)
        .map { id =>
          (id,
            "name_" + id,
            "city_" + id,
            10000.00 * id,
            BigDecimal.valueOf(0.01),
            80.01,
            "1990-01-01",
            "2010-01-01 10:01:01",
            "2010-01-01 10:01:01",
            id.toString() + ("abc" * 12000),
            "school_" + id + "\002school_" + id + id + "\001" + id)
        }
      spark.createDataFrame(csvRDD).toDF("id", "name", "city", "salary", "tax", "percent", "birthday", "register", "updated", "longstr", "file")
    }

    csvDataDF.write
      .option("header", "false")
      .mode(saveMode)
      .csv(csvDirPath)
  }

  def createFileStreamingThread(
      spark: SparkSession,
      carbonTable: CarbonTable,
      csvDataDir: String,
      intervalSecond: Int,
      tableIdentifier: TableIdentifier): Thread = {
    new Thread() {
      override def run(): Unit = {
        var qry: StreamingQuery = null
        try {
          val readSocketDF = spark.readStream.text(csvDataDir)

          // Write data from socket stream to carbondata file
          qry = readSocketDF.writeStream
            .format("carbondata")
            .trigger(ProcessingTime(s"${ intervalSecond } seconds"))
            .option("checkpointLocation",
              CarbonTablePath.getStreamingCheckpointDir(carbonTable.getTablePath))
            .option("dbName", tableIdentifier.database.get)
            .option("tableName", tableIdentifier.table)
            .option("timestampformat", CarbonCommonConstants.CARBON_TIMESTAMP_DEFAULT_FORMAT)
            .option(CarbonStreamParser.CARBON_STREAM_PARSER,
              CarbonStreamParser.CARBON_STREAM_PARSER_CSV)
            .start()

          qry.awaitTermination()
        } catch {
          case _: InterruptedException =>
            // scalastyle:off println
            println("Done reading and writing streaming data")
            // scalastyle:on println
        } finally {
          if (qry != null) {
            qry.stop()
          }
        }
      }
    }
  }

  def createTable(tableName: String, streaming: Boolean, withBatchLoad: Boolean): Unit = {
    sql(
      s"""
         | CREATE TABLE streaming_longstr.$tableName(
         | id INT,
         | name STRING,
         | city STRING,
         | salary FLOAT,
         | tax DECIMAL(8,2),
         | percent double,
         | birthday DATE,
         | register TIMESTAMP,
         | updated TIMESTAMP,
         | longstr STRING
         | )
         | STORED AS carbondata
         | TBLPROPERTIES(${if (streaming) "'streaming'='true', " else "" }
         | 'sort_columns'='name', 'LONG_STRING_COLUMNS'='longstr')
         | """.stripMargin)

    if (withBatchLoad) {
      // batch loading 5 rows
      executeBatchLoad(tableName)
    }
  }

  def createTableWithComplexType(
      tableName: String,
      streaming: Boolean,
      withBatchLoad: Boolean): Unit = {
    sql(
      s"""
         | CREATE TABLE streaming_longstr.$tableName(
         | id INT,
         | name STRING,
         | city STRING,
         | salary FLOAT,
         | tax DECIMAL(8,2),
         | percent double,
         | birthday DATE,
         | register TIMESTAMP,
         | updated TIMESTAMP,
         | longstr STRING,
         | file struct<school:array<string>, age:int>
         | )
         | STORED AS carbondata
         | TBLPROPERTIES(${if (streaming) "'streaming'='true', " else "" }
         | 'sort_columns'='name', 'LONG_STRING_COLUMNS'='longstr')
         | """.stripMargin)

    if (withBatchLoad) {
      // batch loading 5 rows
      executeBatchLoad(tableName)
    }
  }

  def executeBatchLoad(tableName: String): Unit = {
    sql(
      s"LOAD DATA LOCAL INPATH '$dataFilePath' INTO TABLE streaming_longstr.$tableName OPTIONS" +
      "('HEADER'='true','COMPLEX_DELIMITER_LEVEL_1'='$', 'COMPLEX_DELIMITER_LEVEL_2'=':')")
  }

  private def wrap(array: Array[String]) = {
    new mutable.WrappedArray.ofRef(array)
  }

  /**
   * get a ServerSocket
   * if the address was already used, it will retry to use new port number.
   *
   * @return ServerSocket
   */
  def getServerSocket(): ServerSocket = {
    var port = 7071
    var serverSocket: ServerSocket = null
    var retry = false
    do {
      try {
        retry = false
        serverSocket = new ServerSocket(port)
      } catch {
        case ex: BindException =>
          retry = true
          port = port + 2
          if (port >= 65535) {
            throw ex
          }
      }
    } while (retry)
    serverSocket
  }
  // scalastyle:on lineLength
}
