| /* |
| * 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.carbondata.presto.integrationtest |
| |
| import java.io.{BufferedInputStream, File, FileInputStream} |
| import java.sql.SQLException |
| import java.util |
| |
| import org.apache.commons.codec.binary.Hex |
| import org.apache.commons.io.FileUtils |
| import org.apache.commons.lang.RandomStringUtils |
| import org.scalatest.{BeforeAndAfterAll, BeforeAndAfterEach, FunSuiteLike} |
| |
| import org.apache.carbondata.common.logging.LogServiceFactory |
| import org.apache.carbondata.core.constants.CarbonCommonConstants |
| import org.apache.carbondata.core.datastore.filesystem.CarbonFile |
| import org.apache.carbondata.core.datastore.impl.FileFactory |
| import org.apache.carbondata.core.metadata.datatype.{DataTypes, Field, StructField} |
| import org.apache.carbondata.core.util.{CarbonProperties, CarbonUtil} |
| import org.apache.carbondata.presto.server.{PrestoServer, PrestoTestUtil} |
| import org.apache.carbondata.sdk.file.{CarbonWriter, Schema} |
| |
| class PrestoTestNonTransactionalTableFiles |
| extends FunSuiteLike with BeforeAndAfterAll with BeforeAndAfterEach { |
| |
| private val logger = LogServiceFactory |
| .getLogService(classOf[PrestoTestNonTransactionalTableFiles].getCanonicalName) |
| |
| private val rootPath = new File(this.getClass.getResource("/").getPath |
| + "../../../..").getCanonicalPath |
| private val storePath = s"$rootPath/integration/presto/target/store" |
| private val writerPath = storePath + "/sdk_output/files" |
| private val writerPathBinary = storePath + "/sdk_output/files1" |
| private val prestoServer = new PrestoServer |
| private var varcharString = new String |
| |
| override def beforeAll: Unit = { |
| CarbonProperties.getInstance().addProperty(CarbonCommonConstants.CARBON_WRITTEN_BY_APPNAME, |
| "Presto") |
| val map = new util.HashMap[String, String]() |
| map.put("hive.metastore", "file") |
| map.put("hive.metastore.catalog.dir", s"file://$storePath") |
| prestoServer.startServer("sdk_output", map) |
| prestoServer.execute("drop schema if exists sdk_output") |
| prestoServer.execute("create schema sdk_output") |
| } |
| |
| override def afterAll(): Unit = { |
| prestoServer.stopServer() |
| CarbonUtil.deleteFoldersAndFiles(FileFactory.getCarbonFile(storePath)) |
| } |
| |
| def buildTestDataSingleFile(): Any = { |
| FileUtils.deleteDirectory(new File(writerPath)) |
| createTable |
| |
| buildTestData(3, null, true) |
| } |
| |
| def buildStructData(): Any = { |
| FileUtils.deleteDirectory(new File(writerPath)) |
| createTable |
| buildTestData(3, null, true) |
| } |
| |
| def buildTestDataMultipleFiles(): Any = { |
| FileUtils.deleteDirectory(new File(writerPath)) |
| createTable |
| buildTestData(1000000, null, false) |
| } |
| |
| private def createTable = { |
| prestoServer.execute("drop table if exists sdk_output.files") |
| prestoServer |
| .execute( |
| "create table sdk_output.files(name varchar, age int, id tinyint, height double, salary " + |
| "real, address varchar) with" + |
| "(format='CARBON') ") |
| } |
| |
| private def createTableBinary = { |
| prestoServer.execute("drop table if exists sdk_output.files1") |
| prestoServer.execute( |
| "create table sdk_output.files1(name boolean, age int, id varbinary, height double, salary " + |
| "real) with(format='CARBON') ") |
| } |
| |
| def buildTestData(rows: Int, options: util.Map[String, String], varcharDataGen: Boolean): Any = { |
| buildTestData(rows, options, List("name"), varcharDataGen) |
| } |
| |
| // prepare sdk writer output |
| def buildTestData(rows: Int, |
| options: util.Map[String, String], |
| sortColumns: List[String], |
| varcharDataGen: Boolean): Any = { |
| val schema = new StringBuilder() |
| .append("[ \n") |
| .append(" {\"NaMe\":\"string\"},\n") |
| .append(" {\"age\":\"int\"},\n") |
| .append(" {\"id\":\"byte\"},\n") |
| .append(" {\"height\":\"double\"},\n") |
| .append(" {\"salary\":\"float\"},\n") |
| .append(" {\"address\":\"varchar\"}\n") |
| .append("]") |
| .toString() |
| |
| // Build Varchar Column data |
| var varcharValue: String = { |
| if (varcharDataGen) { |
| RandomStringUtils.randomAlphabetic(32001) |
| } else { |
| "a" |
| } |
| } |
| |
| varcharString = varcharValue |
| try { |
| val builder = CarbonWriter.builder() |
| val writer = |
| if (options != null) { |
| builder.outputPath(writerPath) |
| .sortBy(sortColumns.toArray) |
| .uniqueIdentifier( |
| System.currentTimeMillis).withBlockSize(2).withLoadOptions(options) |
| .withCsvInput(Schema.parseJson(schema)).writtenBy("TestNonTransactionalCarbonTable") |
| .build() |
| } else { |
| builder.outputPath(writerPath) |
| .sortBy(sortColumns.toArray) |
| .uniqueIdentifier( |
| System.currentTimeMillis).withBlockSize(2) |
| .withCsvInput(Schema.parseJson(schema)).writtenBy("TestNonTransactionalCarbonTable") |
| .build() |
| } |
| var i = 0 |
| while (i < rows) { |
| if ((options != null) && (i < 3)) { |
| // writing a bad record |
| writer |
| .write(Array[String]("robot" + i, |
| String.valueOf(i), |
| String.valueOf(i.toDouble / 2), |
| "robot", |
| String.valueOf(i.toFloat / 2), |
| String.valueOf(varcharValue))) |
| } else { |
| writer |
| .write(Array[String]("robot" + i, |
| String.valueOf(i), |
| String.valueOf(i % 128), |
| String.valueOf(i.toDouble / 2), |
| String.valueOf(i.toFloat / 2), |
| String.valueOf(varcharValue))) |
| } |
| i += 1 |
| } |
| if (options != null) { |
| // Keep one valid record. else carbon data file will not generate |
| writer |
| .write(Array[String]("robot" + i, |
| String.valueOf(i), |
| String.valueOf(i), |
| String.valueOf(i.toDouble / 2), |
| String.valueOf(i.toFloat / 2), |
| String.valueOf(varcharValue))) |
| } |
| writer.close() |
| } catch { |
| case ex: Throwable => throw new RuntimeException(ex) |
| } |
| } |
| |
| // prepare sdk writer output with other schema |
| def buildTestDataOtherDataType(rows: Int, sortColumns: Array[String], path : String): Any = { |
| val fields: Array[Field] = new Array[Field](5) |
| // same column name, but name as boolean type |
| fields(0) = new Field("name", DataTypes.BOOLEAN) |
| fields(1) = new Field("age", DataTypes.INT) |
| fields(2) = new Field("id", DataTypes.BINARY) |
| fields(3) = new Field("height", DataTypes.DOUBLE) |
| fields(4) = new Field("salary", DataTypes.FLOAT) |
| |
| val imagePath = rootPath + "/sdk/sdk/src/test/resources/image/carbondatalogo.jpg" |
| try { |
| var i = 0 |
| val bis = new BufferedInputStream(new FileInputStream(imagePath)) |
| var hexValue: Array[Char] = null |
| val originBinary = new Array[Byte](bis.available) |
| while (bis.read(originBinary) != -1) { |
| hexValue = Hex.encodeHex(originBinary) |
| } |
| bis.close() |
| val binaryValue = String.valueOf(hexValue) |
| val builder = CarbonWriter.builder() |
| val writer = |
| builder.outputPath(path) |
| .uniqueIdentifier(System.currentTimeMillis()).withBlockSize(2).sortBy(sortColumns) |
| .withCsvInput(new Schema(fields)).writtenBy("TestNonTransactionalCarbonTable").build() |
| while (i < rows) { |
| writer |
| .write(Array[String]("true", |
| String.valueOf(i), |
| binaryValue, |
| String.valueOf(i.toDouble / 2), |
| String.valueOf(i.toFloat / 2))) |
| i += 1 |
| } |
| writer |
| .write(Array[String]("true", |
| String.valueOf(i), |
| String.valueOf("abc"), |
| String.valueOf(i.toDouble / 2), |
| String.valueOf(i.toFloat / 2))) |
| writer.close() |
| } catch { |
| case ex: Throwable => throw new RuntimeException(ex) |
| } |
| } |
| |
| def cleanTestData(): Unit = { |
| FileUtils.deleteDirectory(new File(writerPath)) |
| } |
| |
| def deleteFile(path: String, extension: String): Unit = { |
| val file: CarbonFile = FileFactory.getCarbonFile(path) |
| for (eachDir <- file.listFiles) { |
| if (!eachDir.isDirectory) { |
| if (eachDir.getName.endsWith(extension)) { |
| CarbonUtil.deleteFoldersAndFilesSilent(eachDir) |
| } |
| } else { |
| deleteFile(eachDir.getPath, extension) |
| } |
| } |
| } |
| |
| test("test show schemas") { |
| buildTestDataSingleFile() |
| val actualResult: List[Map[String, Any]] = prestoServer |
| .executeQuery("show schemas ") |
| assert(actualResult |
| .equals(List(Map("Schema" -> "information_schema"), |
| Map("Schema" -> "sdk_output")))) |
| cleanTestData() |
| } |
| |
| test("test show tables") { |
| buildTestDataSingleFile() |
| val actualResult: List[Map[String, Any]] = prestoServer |
| .executeQuery("show tables") |
| assert(actualResult |
| .equals(List(Map("Table" -> "files")))) |
| cleanTestData() |
| } |
| |
| test("test read sdk output files") { |
| buildTestDataSingleFile() |
| val actualResult: List[Map[String, Any]] = prestoServer |
| .executeQuery("SELECT COUNT(*) AS RESULT FROM files ") |
| val expectedResult: List[Map[String, Any]] = List(Map("RESULT" -> 3)) |
| assert(actualResult.equals(expectedResult)) |
| |
| val FloatFilterResult: List[Map[String, Any]] = prestoServer |
| .executeQuery("SELECT * FROM files where salary > 0.0") |
| assert(FloatFilterResult.length == 2 ) |
| |
| val ByteFilterResult: List[Map[String, Any]] = prestoServer |
| .executeQuery("SELECT * FROM files where id = 2") |
| assert(ByteFilterResult.length == 1 ) |
| cleanTestData() |
| } |
| |
| test("test read multiple carbondata and index files") { |
| buildTestDataMultipleFiles() |
| val actualResult: List[Map[String, Any]] = prestoServer |
| .executeQuery("SELECT COUNT(*) AS RESULT FROM files ") |
| val expectedResult: List[Map[String, Any]] = List(Map("RESULT" -> 1000000)) |
| assert(actualResult.equals(expectedResult)) |
| cleanTestData() |
| } |
| |
| test("test reading different schema") { |
| buildTestDataSingleFile() |
| buildTestDataOtherDataType(3, null, writerPath) |
| val exception = |
| intercept[SQLException] { |
| val actualResult: List[Map[String, Any]] = prestoServer |
| .executeQuery("select count(*) as RESULT from files ") |
| } |
| assert(exception.getMessage() |
| .contains("All common columns present in the files doesn't have same datatype")) |
| cleanTestData() |
| } |
| |
| test("test reading binary") { |
| FileUtils.deleteDirectory(new File(writerPathBinary)) |
| createTableBinary |
| buildTestDataOtherDataType(3, null, writerPathBinary) |
| val actualResult: List[Map[String, Any]] = prestoServer |
| .executeQuery("select id from files1 ") |
| assert(actualResult.size == 4) |
| // check the binary byte Array size, as original hex encoded image byte array size is 118198 |
| assert(actualResult.head("id").asInstanceOf[Array[Byte]].length == 118198) |
| // validate some initial bytes |
| assert(actualResult.head("id").asInstanceOf[Array[Byte]](0) == 56) |
| assert(actualResult.head("id").asInstanceOf[Array[Byte]](1) == 57) |
| assert(actualResult.head("id").asInstanceOf[Array[Byte]](2) == 53) |
| assert(actualResult.head("id").asInstanceOf[Array[Byte]](3) == 48) |
| |
| val binaryFilterResult: List[Map[String, Any]] = prestoServer |
| .executeQuery("select id from files1 where id = cast('abc' as varbinary)"); |
| assert(binaryFilterResult.length == 1) |
| FileUtils.deleteDirectory(new File(writerPathBinary)) |
| } |
| |
| test("test reading without carbon index file") { |
| buildTestDataSingleFile() |
| deleteFile(writerPath, CarbonCommonConstants.UPDATE_INDEX_FILE_EXT) |
| val exception = |
| intercept[SQLException] { |
| val actualResult: List[Map[String, Any]] = prestoServer |
| .executeQuery("select * from files ") |
| } |
| assert(exception.getMessage() |
| .contains("No Index files are present in the table location")) |
| cleanTestData() |
| } |
| |
| test("test select all columns") { |
| buildTestDataSingleFile() |
| val actualResult: List[Map[String, Any]] = prestoServer |
| .executeQuery("select * from files ") |
| val expectedResult: List[Map[String, Any]] = List(Map( |
| "name" -> "robot0", |
| "height" -> 0.0, |
| "age" -> 0, |
| "salary" -> 0.0, |
| "id" -> 0, |
| "address" -> varcharString), |
| Map("name" -> "robot1", |
| "height" -> 0.5, |
| "age" -> 1, |
| "salary" -> 0.5, |
| "id" -> 1, |
| "address" -> varcharString), |
| Map("name" -> "robot2", |
| "height" -> 1.0, |
| "age" -> 2, |
| "salary" -> 1.0, |
| "id" -> 2, |
| "address" -> varcharString)) |
| assert(actualResult.toString() equals expectedResult.toString()) |
| } |
| |
| test("Test for query on Varchar columns") { |
| buildTestDataSingleFile() |
| val actualRes: List[Map[String, Any]] = prestoServer. |
| executeQuery("select max(length(address)) from files") |
| val expectedRes: List[Map[String, Any]] = List(Map("_col0" -> 32001)) |
| assert(actualRes.toString() equals expectedRes.toString()) |
| } |
| |
| test("test struct of primitive type") { |
| import scala.collection.JavaConverters._ |
| val writerPathComplex = storePath + "/sdk_output/files4" |
| FileUtils.deleteDirectory(new File(writerPathComplex)) |
| prestoServer.execute("drop table if exists sdk_output.files4") |
| prestoServer.execute( |
| "create table sdk_output.files4(stringField varchar, structField ROW(byteField tinyint, " + |
| "shortField SMALLINT, intField Integer, longField BIGINT, floatField real, doubleField " + |
| "DOUBLE, binaryField varbinary, dateField date, timeStampField timestamp, booleanField " + |
| "boolean, longStringField varchar, decimalField decimal(8,2), stringChildField varchar)) " + |
| "with(format='CARBON') ") |
| |
| val imagePath = rootPath + "/sdk/sdk/src/test/resources/image/carbondatalogo.jpg" |
| val bis = new BufferedInputStream(new FileInputStream(imagePath)) |
| var hexValue: Array[Char] = null |
| val originBinary = new Array[Byte](bis.available) |
| while (bis.read(originBinary) != -1) { |
| hexValue = Hex.encodeHex(originBinary) |
| } |
| bis.close() |
| val binaryValue = String.valueOf(hexValue) |
| |
| val longChar = RandomStringUtils.randomAlphabetic(33000) |
| |
| val fields = List(new StructField("byteField", DataTypes.BYTE), |
| new StructField("shortField", DataTypes.SHORT), |
| new StructField("intField", DataTypes.INT), |
| new StructField("longField", DataTypes.LONG), |
| new StructField("floatField", DataTypes.FLOAT), |
| new StructField("doubleField", DataTypes.DOUBLE), |
| new StructField("binaryField", DataTypes.BINARY), |
| new StructField("dateField", DataTypes.DATE), |
| new StructField("timeStampField", DataTypes.TIMESTAMP), |
| new StructField("booleanField", DataTypes.BOOLEAN), |
| new StructField("longStringField", DataTypes.VARCHAR), |
| new StructField("decimalField", DataTypes.createDecimalType(8, 2)), |
| new StructField("stringChildField", DataTypes.STRING)) |
| val structType = Array(new Field("stringField", DataTypes.STRING), new Field |
| ("structField", "struct", fields.asJava)) |
| |
| try { |
| val options: util.Map[String, String] = |
| Map("bAd_RECords_action" -> "FORCE", "quotechar" -> "\"").asJava |
| val builder = CarbonWriter.builder() |
| val writer = |
| builder.outputPath(writerPathComplex) |
| .uniqueIdentifier(System.nanoTime()) |
| .withLoadOptions(options) |
| .withBlockSize(2) |
| .enableLocalDictionary(false) |
| .withCsvInput(new Schema(structType)).writtenBy("presto").build() |
| |
| val array1 = Array[String]("row1", null, null, null, null, null, null, null, null, null, |
| null, null, null, null) |
| |
| val array2 = Array[String]("row2", "5" + "\001" + "5" + "\001" + "5" + "\001" + "5" + |
| "\001" + "5.512" + "\001" + "5.512" + "\001" + |
| binaryValue + "\001" + "2019-03-02" + "\001" + |
| "2019-02-12 03:03:34" + "\001" + "true" + "\001" + |
| longChar + "\001" + "-2.2" + "\001" + "stringName") |
| writer.write(array1) |
| writer.write(array2) |
| writer.close() |
| } catch { |
| case ex: Exception => throw new RuntimeException(ex) |
| case _: Throwable => None |
| } |
| val actualResult: List[Map[String, Any]] = prestoServer |
| .executeQuery("select * from files4 ") |
| assert(actualResult.size == 2) |
| |
| |
| for(i <- 0 to 1) { |
| val row = actualResult(i)("stringfield") |
| val result = actualResult(i)("structfield").asInstanceOf[java.util.Map[String, Any]] |
| if(row == "row1") { assert(result.get("bytefield") == null) |
| assert(result.get("shortfield") == null) |
| assert(result.get("intfield") == null) |
| assert(result.get("longfield") == null) |
| assert(result.get("floatfield") == null) |
| assert(result.get("doublefield") == null) |
| assert(result.get("binaryfield") == null) |
| assert(result.get("datefield") == null) |
| assert(result.get("timestampfield") == null) |
| assert(result.get("booleanfield") == null) |
| assert(result.get("longstringfield") == null) |
| assert(result.get("decimalfield") == null) |
| assert(result.get("stringchildfield") == null) |
| } else { |
| assert(result.get("bytefield") == 5) |
| assert(result.get("shortfield") == 5) |
| assert(result.get("intfield") == 5) |
| assert(result.get("longfield") == 5L) |
| assert(result.get("floatfield") == 5.512f) |
| assert(result.get("doublefield") == 5.512) |
| assert((result.get("binaryfield").asInstanceOf[Array[Byte]]).length == 118198) |
| assert(result.get("datefield") == "2019-03-02") |
| assert(result.get("timestampfield") == "2019-02-12 03:03:34.000") |
| assert(result.get("booleanfield") == true) |
| assert(result.get("longstringfield") == longChar) |
| assert(result.get("decimalfield") == "-2.20") |
| assert(result.get("stringchildfield") == "stringName") |
| } |
| } |
| FileUtils.deleteDirectory(new File(writerPathComplex)) |
| } |
| |
| test("test struct of date type with huge data") { |
| import scala.collection.JavaConverters._ |
| val writerPathComplex = storePath + "/sdk_output/files2" |
| FileUtils.deleteDirectory(new File(writerPathComplex)) |
| prestoServer.execute("drop table if exists sdk_output.files2") |
| prestoServer |
| .execute( |
| "create table sdk_output.files2(structField ROW(dateField date)) with(format='CARBON') ") |
| val fields = List( |
| new StructField("dateField", DataTypes.DATE)) |
| val structType = Array(new Field("structField", "struct", fields.asJava)) |
| try { |
| val builder = CarbonWriter.builder() |
| val writer = |
| builder.outputPath(writerPathComplex) |
| .uniqueIdentifier(System.nanoTime()).withBlockSize(2).enableLocalDictionary(false) |
| .withCsvInput(new Schema(structType)).writtenBy("presto").build() |
| var i = 0 |
| while (i < 100000) { |
| val array = Array[String]("2019-03-02") |
| writer.write(array) |
| i += 1 |
| } |
| writer.close() |
| } catch { |
| case ex: Exception => throw new RuntimeException(ex) |
| case _: Throwable => None |
| } |
| val actualResult: List[Map[String, Any]] = prestoServer |
| .executeQuery("select structField from files2 ") |
| assert(actualResult.size == 100000) |
| FileUtils.deleteDirectory(new File(writerPathComplex)) |
| } |
| |
| test("test Array of primitive type") { |
| val writerPathComplex = storePath + "/sdk_output/files5" |
| import scala.collection.JavaConverters._ |
| FileUtils.deleteDirectory(new File(writerPathComplex)) |
| prestoServer.execute("drop table if exists sdk_output.files5") |
| prestoServer.execute( |
| "create table sdk_output.files5(arrayByte ARRAY(tinyint), arrayShort ARRAY(smallint), " + |
| "arrayInt ARRAY(int), arrayLong ARRAY(bigint), arrayFloat ARRAY(real), arrayDouble ARRAY" + |
| "(double), arrayBinary ARRAY(varbinary), arrayDate ARRAY(date), arrayTimestamp ARRAY" + |
| "(timestamp), arrayBoolean ARRAY(boolean), arrayVarchar ARRAY(varchar), arrayDecimal ARRAY" + |
| "(decimal(8,2)), arrayString ARRAY(varchar), stringField varchar ) with(format='CARBON') ") |
| |
| val imagePath = rootPath + "/sdk/sdk/src/test/resources/image/carbondatalogo.jpg" |
| val bis = new BufferedInputStream(new FileInputStream(imagePath)) |
| var hexValue: Array[Char] = null |
| val originBinary = new Array[Byte](bis.available) |
| while (bis.read(originBinary) != -1) { |
| hexValue = Hex.encodeHex(originBinary) |
| } |
| bis.close() |
| val binaryValue = String.valueOf(hexValue) |
| |
| val longChar = RandomStringUtils.randomAlphabetic(33000) |
| |
| val fields1 = List(new StructField("byteField", DataTypes.BYTE)) |
| val structType1 = new Field("arrayByte", "array", fields1.asJava) |
| val fields2 = List(new StructField("shortField", DataTypes.SHORT)) |
| val structType2 = new Field("arrayShort", "array", fields2.asJava) |
| val fields3 = List(new StructField("intField", DataTypes.INT)) |
| val structType3 = new Field("arrayInt", "array", fields3.asJava) |
| val fields4 = List(new StructField("longField", DataTypes.LONG)) |
| val structType4 = new Field("arrayLong", "array", fields4.asJava) |
| val fields5 = List(new StructField("floatField", DataTypes.FLOAT)) |
| val structType5 = new Field("arrayFloat", "array", fields5.asJava) |
| val fields6 = List(new StructField("DoubleField", DataTypes.DOUBLE)) |
| val structType6 = new Field("arrayDouble", "array", fields6.asJava) |
| val fields7 = List(new StructField("binaryField", DataTypes.BINARY)) |
| val structType7 = new Field("arrayBinary", "array", fields7.asJava) |
| val fields8 = List(new StructField("dateField", DataTypes.DATE)) |
| val structType8 = new Field("arrayDate", "array", fields8.asJava) |
| val fields9 = List(new StructField("timestampField", DataTypes.TIMESTAMP)) |
| val structType9 = new Field("arrayTimestamp", "array", fields9.asJava) |
| val fields10 = List(new StructField("booleanField", DataTypes.BOOLEAN)) |
| val structType10 = new Field("arrayBoolean", "array", fields10.asJava) |
| val fields11 = List(new StructField("varcharField", DataTypes.VARCHAR)) |
| val structType11 = new Field("arrayVarchar", "array", fields11.asJava) |
| val fields12 = List(new StructField("decimalField", DataTypes.createDecimalType(8, 2))) |
| val structType12 = new Field("arrayDecimal", "array", fields12.asJava) |
| val fields13 = List(new StructField("stringField", DataTypes.STRING)) |
| val structType13 = new Field("arrayString", "array", fields13.asJava) |
| val structType14 = new Field("stringField", DataTypes.STRING) |
| |
| try { |
| val options: util.Map[String, String] = |
| Map("bAd_RECords_action" -> "FORCE", "quotechar" -> "\"").asJava |
| val builder = CarbonWriter.builder() |
| val writer = |
| builder.outputPath(writerPathComplex).withLoadOptions(options) |
| .uniqueIdentifier(System.nanoTime()).withBlockSize(2).enableLocalDictionary(false) |
| .withCsvInput(new Schema(Array[Field]( |
| structType1, structType2, structType3, structType4, structType5, structType6, |
| structType7, structType8, structType9, structType10, structType11, structType12, |
| structType13, structType14))).writtenBy("presto").build() |
| |
| var array = Array[String](null, null, null, null, null, null, null, null, null, null, |
| null, null, null, "row1") |
| writer.write(array) |
| array = Array[String]("3" + "\001" + "5" + "\001" + "4", |
| "4" + "\001" + "5" + "\001" + "6", |
| "4", |
| "2" + "\001" + "59999999" + "\001" + "99999999999", |
| "5.4646" + "\001" + "5.55" + "\001" + "0.055", |
| "5.46464646464" + "\001" + "5.55" + "\001" + "0.055", |
| binaryValue, |
| "2019-03-02" + "\001" + "2020-03-02" + "\001" + "2021-04-02", |
| "2019-02-12 03:03:34" + "\001" +"2020-02-12 03:03:34" + "\001" + "2021-03-12 03:03:34", |
| "true" + "\001" + "false", |
| longChar, |
| "999.232323" + "\001" + "0.1234", |
| "japan" + "\001" + "china" + "\001" + "iceland", |
| "row2") |
| writer.write(array) |
| writer.close() |
| } catch { |
| case e: Exception => |
| assert(false) |
| } |
| val actualResult: List[Map[String, Any]] = prestoServer |
| .executeQuery("select * from files5 ") |
| |
| assert(actualResult.size == 2) |
| PrestoTestUtil.validateArrayOfPrimitiveTypeData(actualResult, longChar) |
| FileUtils.deleteDirectory(new File(writerPathComplex)) |
| } |
| |
| test("test Array of date type with huge data") { |
| val writerPathComplex = storePath + "/sdk_output/files6" |
| import scala.collection.JavaConverters._ |
| FileUtils.deleteDirectory(new File(writerPathComplex)) |
| prestoServer.execute("drop table if exists sdk_output.files6") |
| prestoServer |
| .execute( |
| "create table sdk_output.files6(arrayDate ARRAY(date)) with(format='CARBON') ") |
| val fields8 = List(new StructField("intField", DataTypes.DATE)) |
| val structType8 = new Field("arrayDate", "array", fields8.asJava) |
| try { |
| val builder = CarbonWriter.builder() |
| val writer = |
| builder.outputPath(writerPathComplex) |
| .uniqueIdentifier(System.nanoTime()).withBlockSize(2).enableLocalDictionary(false) |
| .withCsvInput(new Schema(Array[Field](structType8))).writtenBy("presto").build() |
| |
| var i = 0 |
| while (i < 50000) { |
| val array = Array[String]("2019-03-02" + "\001" + "2020-03-02" + "\001" + "2021-04-02") |
| writer.write(array) |
| val array1 = Array[String]("2021-04-02") |
| writer.write(array1) |
| i += 1 |
| } |
| writer.close() |
| } catch { |
| case e: Exception => |
| assert(false) |
| } |
| val actualResult: List[Map[String, Any]] = prestoServer |
| .executeQuery("select * from files6 ") |
| assert(actualResult.size == 100 * 1000) |
| FileUtils.deleteDirectory(new File(writerPathComplex)) |
| } |
| |
| test("test Array with local dictionary") { |
| val writerPathComplex = storePath + "/sdk_output/files7" |
| import scala.collection.JavaConverters._ |
| FileUtils.deleteDirectory(new File(writerPathComplex)) |
| prestoServer.execute("drop table if exists sdk_output.files7") |
| prestoServer |
| .execute( |
| "create table sdk_output.files7(arrayString ARRAY(varchar), arrayDate ARRAY(DATE), " + |
| "arrayVarchar ARRAY(varchar), stringField varchar ) with(format='CARBON') ") |
| |
| val field1 = List(new StructField("stringField", DataTypes.STRING)) |
| val structType1 = new Field("arrayString", "array", field1.asJava) |
| val field2 = List(new StructField("dateField", DataTypes.DATE)) |
| val structType2 = new Field("arrayDate", "array", field2.asJava) |
| val fields3 = List(new StructField("varcharField", DataTypes.VARCHAR)) |
| val structType3 = new Field("arrayVarchar", "array", fields3.asJava) |
| val structType4 = new Field("stringField", DataTypes.STRING) |
| |
| val longChar = RandomStringUtils.randomAlphabetic(33000) |
| |
| try { |
| val options: util.Map[String, String] = Map("bAd_RECords_action" -> "FORCE", |
| "quotechar" -> "\"").asJava |
| val builder = CarbonWriter.builder() |
| val writer = |
| builder.outputPath(writerPathComplex).withLoadOptions(options) |
| .uniqueIdentifier(System.nanoTime()).withBlockSize(2).enableLocalDictionary(true) |
| .withCsvInput(new Schema(Array[Field](structType1, |
| structType2, |
| structType3, |
| structType4))).writtenBy("presto").build() |
| |
| var array = Array[String](null, |
| null, |
| null, |
| "row1") |
| writer.write(array) |
| array = Array[String]("India" + "\001" + "Japan" + "\001" + "India", |
| "2019-03-02" + "\001" + "2020-03-02", |
| longChar, |
| "row2") |
| writer.write(array) |
| array = Array[String]( |
| "Iceland", |
| "2019-03-02" + "\001" + "2020-03-02" + "\001" + "2021-04-02" + "\001" + "2021-04-03" + |
| "\001" + "2021-04-02", |
| longChar, |
| "row3") |
| writer.write(array) |
| |
| writer.close() |
| } catch { |
| case e: Exception => |
| assert(false) |
| } |
| val actualResult: List[Map[String, Any]] = prestoServer |
| .executeQuery("select * from files7 ") |
| PrestoTestUtil.validateArrayOfPrimitiveTypeDataWithLocalDict(actualResult, longChar) |
| FileUtils.deleteDirectory(new File(writerPathComplex)) |
| } |
| |
| test("test Struct with local dictionary") { |
| import scala.collection.JavaConverters._ |
| val writerPathComplex = storePath + "/sdk_output/files8" |
| FileUtils.deleteDirectory(new File(writerPathComplex)) |
| prestoServer.execute("drop table if exists sdk_output.files8") |
| prestoServer |
| .execute( |
| "create table sdk_output.files8(stringField varchar, structField ROW(stringChildField " + |
| "varchar, dateField date, longStringField varchar)) with(format='CARBON') ") |
| val longChar = RandomStringUtils.randomAlphabetic(33000) |
| |
| val fields = List(new StructField("stringChildField", DataTypes.STRING), |
| new StructField("dateField", DataTypes.DATE), |
| new StructField("longStringField", DataTypes.VARCHAR) |
| ) |
| val structType = Array(new Field("stringField", DataTypes.STRING), new Field |
| ("structField", "struct", fields.asJava)) |
| try { |
| val options: util.Map[String, String] = Map("bAd_RECords_action" -> "FORCE", |
| "quotechar" -> "\"").asJava |
| val builder = CarbonWriter.builder() |
| val writer = |
| builder.outputPath(writerPathComplex) |
| .uniqueIdentifier(System.nanoTime()) |
| .withLoadOptions(options) |
| .withBlockSize(2) |
| .enableLocalDictionary(true) |
| .withCsvInput(new Schema(structType)) |
| .writtenBy("presto") |
| .build() |
| |
| val array1 = Array[String]("row1", |
| null, |
| null, |
| null) |
| val array2 = Array[String]("row2", "local dictionary" |
| + "\001" + "2019-03-02" |
| + "\001" + longChar) |
| writer.write(array1) |
| writer.write(array2) |
| writer.close() |
| } catch { |
| case ex: Exception => throw new RuntimeException(ex) |
| case _: Throwable => None |
| } |
| |
| val actualResult: List[Map[String, Any]] = prestoServer |
| .executeQuery("select * from files8 ") |
| assert(actualResult.size == 2) |
| |
| for (i <- 0 to 1) { |
| val row = actualResult(i)("stringfield") |
| val result = actualResult(i)("structfield").asInstanceOf[java.util.Map[String, Any]] |
| if (row == "row1") { |
| assert(result.get("stringchildfield") == null) |
| assert(result.get("datefield") == null) |
| assert(result.get("longStringField") == null) |
| } |
| else if (row == "row2") { |
| assert(result.get("stringchildfield") == "local dictionary") |
| assert(result.get("datefield") == "2019-03-02") |
| assert(result.get("longstringfield") == longChar) |
| } |
| } |
| FileUtils.deleteDirectory(new File(writerPathComplex)) |
| |
| } |
| |
| test("test Array of varchar type with huge data enabling local dictionary") { |
| val writerPathComplex = storePath + "/sdk_output/files9" |
| import scala.collection.JavaConverters._ |
| FileUtils.deleteDirectory(new File(writerPathComplex)) |
| prestoServer.execute("drop table if exists sdk_output.files9") |
| prestoServer |
| .execute( |
| "create table sdk_output.files9(arrayString ARRAY(varchar)) with(format='CARBON') ") |
| val fields8 = List(new StructField("intField", DataTypes.STRING)) |
| val structType8 = new Field("arrayString", "array", fields8.asJava) |
| try { |
| val builder = CarbonWriter.builder() |
| val writer = |
| builder.outputPath(writerPathComplex) |
| .uniqueIdentifier(System.nanoTime()).withBlockSize(2).enableLocalDictionary(true) |
| .withCsvInput(new Schema(Array[Field](structType8))).writtenBy("presto").build() |
| |
| var i = 0 |
| while (i < 50000) { |
| val array = Array[String]("India" + "\001" + "China" + "\001" + "Japan") |
| writer.write(array) |
| val array1 = Array[String]("Korea") |
| writer.write(array1) |
| i += 1 |
| } |
| writer.close() |
| } catch { |
| case e: Exception => |
| assert(false) |
| } |
| val actualResult: List[Map[String, Any]] = prestoServer |
| .executeQuery("select * from files9 ") |
| PrestoTestUtil.validateHugeDataForArrayWithLocalDict(actualResult) |
| FileUtils.deleteDirectory(new File(writerPathComplex)) |
| } |
| |
| } |
