-
Notifications
You must be signed in to change notification settings - Fork 702
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
Showing
26 changed files
with
677 additions
and
110 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
262 changes: 262 additions & 0 deletions
262
...rc/test/scala/org/apache/carbondata/spark/testsuite/compaction/TestHybridCompaction.scala
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,262 @@ | ||
| package org.apache.carbondata.spark.testsuite.compaction | ||
|
|
||
| import java.io.{BufferedWriter, File, FileWriter} | ||
|
|
||
| import scala.collection.mutable.ListBuffer | ||
|
|
||
| import au.com.bytecode.opencsv.CSVWriter | ||
| import org.apache.commons.io.FileUtils | ||
| import org.apache.spark.sql.test.util.QueryTest | ||
| import org.junit.Assert | ||
| import org.scalatest.Matchers._ | ||
| import org.scalatest.{BeforeAndAfterAll, BeforeAndAfterEach} | ||
|
|
||
| import org.apache.carbondata.core.constants.CarbonCommonConstants | ||
| import org.apache.carbondata.core.util.CarbonProperties | ||
|
|
||
|
|
||
| class TestHybridCompaction extends QueryTest with BeforeAndAfterEach with BeforeAndAfterAll { | ||
|
|
||
| val rootPath = new File(this.getClass.getResource("/").getPath + "../../../..").getCanonicalPath | ||
|
|
||
| val csvPath1 = | ||
| s"$rootPath/integration/spark-common-test/src/test/resources/compaction/hybridCompaction1.csv" | ||
|
|
||
| val csvPath2 = | ||
| s"$rootPath/integration/spark-common-test/src/test/resources/compaction/hybridCompaction2.csv" | ||
|
|
||
| val tableName = "t1" | ||
|
|
||
|
|
||
| override def beforeAll: Unit = { | ||
| generateCSVFiles() | ||
| CarbonProperties.getInstance() | ||
| .addProperty(CarbonCommonConstants.CARBON_DATE_FORMAT, "MM/dd/yyyy") | ||
| } | ||
|
|
||
|
|
||
| override def afterAll: Unit = { | ||
| deleteCSVFiles() | ||
| } | ||
|
|
||
|
|
||
| override def beforeEach(): Unit = { | ||
| dropTable() | ||
| createTable() | ||
| } | ||
|
|
||
|
|
||
| override def afterEach(): Unit = { | ||
| dropTable() | ||
| } | ||
|
|
||
|
|
||
| def generateCSVFiles(): Unit = { | ||
| val rows1 = new ListBuffer[Array[String]] | ||
| rows1 += Array("seq", "first", "last", "age", "city", "state", "date") | ||
| rows1 += Array("1", "Augusta", "Nichols", "20", "Varasdo", "WA", "07/05/2003") | ||
| rows1 += Array("2", "Luis", "Barnes", "39", "Oroaklim", "MT", "04/05/2048") | ||
| rows1 += Array("3", "Leah", "Guzman", "54", "Culeosa", "KS", "02/23/1983") | ||
| rows1 += Array("4", "Ian", "Ford", "61", "Rufado", "AL", "03/02/1995") | ||
| rows1 += Array("5", "Fanny", "Horton", "37", "Rorlihbem", "CT", "05/12/1987") | ||
| createCSV(rows1, csvPath1) | ||
|
|
||
| val rows2 = new ListBuffer[Array[String]] | ||
| rows2 += Array("seq", "first", "last", "age", "city", "state", "date") | ||
| rows2 += Array("11", "Claudia", "Sullivan", "42", "Dilwuani", "ND", "09/01/2003") | ||
| rows2 += Array("12", "Kate", "Adkins", "54", "Fokafrid", "WA", "10/13/2013") | ||
| rows2 += Array("13", "Eliza", "Lynch", "23", "Bonpige", "ME", "05/02/2015") | ||
| rows2 += Array("14", "Sarah", "Fleming", "60", "Duvugove", "IA", "04/15/2036") | ||
| rows2 += Array("15", "Maude", "Bass", "44", "Ukozedka", "CT", "11/08/1988") | ||
| createCSV(rows2, csvPath2) | ||
| } | ||
|
|
||
|
|
||
| def createCSV(rows: ListBuffer[Array[String]], csvPath: String): Unit = { | ||
| val out = new BufferedWriter(new FileWriter(csvPath)) | ||
| val writer: CSVWriter = new CSVWriter(out) | ||
|
|
||
| for (row <- rows) { | ||
| writer.writeNext(row) | ||
| } | ||
|
|
||
| out.close() | ||
| writer.close() | ||
| } | ||
|
|
||
|
|
||
| def deleteCSVFiles(): Unit = { | ||
| try { | ||
| FileUtils.forceDelete(new File(csvPath1)) | ||
| FileUtils.forceDelete(new File(csvPath2)) | ||
| } | ||
| catch { | ||
| case e: Exception => | ||
| e.printStackTrace() | ||
| Assert.fail(e.getMessage) | ||
| } | ||
| } | ||
|
|
||
|
|
||
| def createTable(): Unit = { | ||
| sql( | ||
| s""" | ||
| | CREATE TABLE $tableName(seq int, first string, last string, | ||
| | age int, city string, state string, date date) | ||
| | STORED BY 'carbondata' | ||
| | TBLPROPERTIES( | ||
| | 'sort_scope'='local_sort', | ||
| | 'sort_columns'='state, age', | ||
| | 'dateformat'='MM/dd/yyyy') | ||
| """.stripMargin) | ||
| } | ||
|
|
||
|
|
||
| def loadUnsortedData(n : Int = 1): Unit = { | ||
| for(_ <- 1 to n) { | ||
| sql( | ||
| s""" | ||
| | LOAD DATA INPATH '$csvPath1' INTO TABLE $tableName | ||
| | OPTIONS ( | ||
| | 'sort_scope'='no_sort')""".stripMargin) | ||
| } | ||
| } | ||
|
|
||
|
|
||
| def loadSortedData(n : Int = 1): Unit = { | ||
| for(_ <- 1 to n) { | ||
| sql( | ||
| s""" | ||
| | LOAD DATA INPATH '$csvPath2' INTO TABLE $tableName | ||
| | OPTIONS ( | ||
| | 'sort_scope'='local_sort')""".stripMargin) | ||
| } | ||
| } | ||
|
|
||
|
|
||
| def dropTable(): Unit = { | ||
| sql(s"DROP TABLE IF EXISTS $tableName") | ||
| } | ||
|
|
||
|
|
||
| test("SORTED LOADS") { | ||
| loadSortedData(2) | ||
| sql(s"ALTER TABLE $tableName COMPACT 'major'") | ||
|
|
||
| val out = sql(s"SELECT state, age FROM $tableName").collect() | ||
| out.map(_.get(0).toString) should | ||
| equal(Array("CT", "CT", "IA", "IA", "ME", "ME", "ND", "ND", "WA", "WA")) | ||
| } | ||
|
|
||
|
|
||
| test("UNSORTED LOADS") { | ||
| loadUnsortedData(2) | ||
| sql(s"ALTER TABLE $tableName COMPACT 'major'") | ||
|
|
||
| val out = sql(s"SELECT state, age FROM $tableName").collect() | ||
| out.map(_.get(0).toString) should | ||
| equal(Array("AL", "AL", "CT", "CT", "KS", "KS", "MT", "MT", "WA", "WA")) | ||
| } | ||
|
|
||
|
|
||
| test("MIXED LOADS") { | ||
| loadSortedData() | ||
| loadUnsortedData() | ||
| sql(s"ALTER TABLE $tableName COMPACT 'major'") | ||
| val out = sql(s"SELECT state, age FROM $tableName").collect() | ||
| out.map(_.get(0).toString) should | ||
| equal(Array("AL", "CT", "CT", "IA", "KS", "ME", "MT", "ND", "WA", "WA")) | ||
| out.map(_.get(1).toString) should | ||
| equal(Array("61", "37", "44", "60", "54", "23", "39", "42", "20", "54")) | ||
| } | ||
|
|
||
|
|
||
| test("INSERT") { | ||
| loadSortedData() | ||
| loadUnsortedData() | ||
| sql( | ||
| s""" | ||
| | INSERT INTO $tableName | ||
| | VALUES('20', 'Naman', 'Rastogi', '23', 'Bengaluru', 'ZZ', '12/28/2018') | ||
| """.stripMargin) | ||
| sql(s"ALTER TABLE $tableName COMPACT 'major'") | ||
|
|
||
| val out = sql(s"SELECT state FROM $tableName").collect() | ||
| out.map(_.get(0).toString) should equal( | ||
| Array("AL", "CT", "CT", "IA", "KS", "ME", "MT", "ND", "WA", "WA", "ZZ")) | ||
| } | ||
|
|
||
|
|
||
| test("UPDATE") { | ||
| loadSortedData() | ||
| loadUnsortedData() | ||
| sql(s"UPDATE $tableName SET (state)=('CT') WHERE seq='13'").collect() | ||
| sql(s"ALTER TABLE $tableName COMPACT 'major'") | ||
|
|
||
| val out = sql(s"SELECT state FROM $tableName WHERE seq='13'").collect() | ||
| out.map(_.get(0).toString) should equal(Array("CT")) | ||
| } | ||
|
|
||
| test("DELETE") { | ||
| loadSortedData() | ||
| loadUnsortedData() | ||
| sql(s"DELETE FROM $tableName WHERE seq='13'").collect() | ||
| sql(s"ALTER TABLE $tableName COMPACT 'major'") | ||
|
|
||
| val out = sql(s"SELECT state FROM $tableName").collect() | ||
| out.map(_.get(0).toString) should equal( | ||
| Array("AL", "CT", "CT", "IA", "KS", "MT", "ND", "WA", "WA")) | ||
| } | ||
|
|
||
|
|
||
| test("RESTRUCTURE TABLE REMOVE COLUMN NOT IN SORT_COLUMNS") { | ||
| loadSortedData() | ||
| loadUnsortedData() | ||
| sql(s"ALTER TABLE $tableName DROP COLUMNS(city)") | ||
| sql(s"ALTER TABLE $tableName COMPACT 'major'") | ||
|
|
||
| val out = sql(s"SELECT age FROM $tableName").collect() | ||
| out.map(_.get(0).toString) should equal( | ||
| Array("61", "37", "44", "60", "54", "23", "39", "42", "20", "54")) | ||
| } | ||
|
|
||
|
|
||
| test("RESTRUCTURE TABLE REMOVE COLUMN IN SORT_COLUMNS") { | ||
| loadSortedData() | ||
| loadUnsortedData() | ||
| sql(s"ALTER TABLE $tableName DROP COLUMNS(state)") | ||
| sql(s"ALTER TABLE $tableName COMPACT 'major'") | ||
|
|
||
| val out = sql(s"SELECT age FROM $tableName").collect() | ||
| out.map(_.get(0).toString) should equal( | ||
| Array("20", "23", "37", "39", "42", "44", "54", "54", "60", "61")) | ||
| } | ||
|
|
||
|
|
||
| test("PREAGG") { | ||
| loadSortedData() | ||
| loadUnsortedData() | ||
| val datamapName = "d1" | ||
| val tableNameDatamapName = tableName + "_" + datamapName | ||
|
|
||
| sql( | ||
| s""" | ||
| | CREATE DATAMAP $datamapName | ||
| | ON TABLE $tableName | ||
| | USING 'preaggregate' | ||
| | AS | ||
| | SELECT AVG(age), state | ||
| | FROM $tableName | ||
| | GROUP BY state | ||
| """.stripMargin) | ||
|
|
||
| loadSortedData() | ||
| loadUnsortedData() | ||
|
|
||
| sql(s"ALTER TABLE $tableName COMPACT 'major'") | ||
| val out = sql(s"SELECT * FROM $tableNameDatamapName").collect() | ||
| out.map(_.get(2).toString) should equal( | ||
| Array("AL", "CT", "IA", "KS", "ME", "MT", "ND", "WA")) | ||
| } | ||
|
|
||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.