Github user manishgupta88 commented on a diff in the pull request:
https://github.com/apache/carbondata/pull/2084#discussion_r176651976
--- Diff: integration/spark2/src/test/scala/org/apache/spark/carbondata/TestStreamingTableOperation.scala ---
@@ -249,6 +251,91 @@ class TestStreamingTableOperation extends QueryTest with BeforeAndAfterAll {
assertResult(exceptedRow)(row)
}
+ test("test preaggregate table creation on streaming table without handoff") {
+ val identifier = new TableIdentifier("agg_table", Option("streaming"))
+ val carbonTable = CarbonEnv.getInstance(spark).carbonMetastore.lookupRelation(identifier)(spark)
+ .asInstanceOf[CarbonRelation].metaData.carbonTable
+ val csvDataDir = new File("target/csvdatanew").getCanonicalPath
+ // streaming ingest 10 rows
+ generateCSVDataFile(spark, idStart = 10, rowNums = 5, csvDataDir)
+ val thread = createFileStreamingThread(spark, carbonTable, csvDataDir, intervalSecond = 1,
+ identifier)
+ thread.start()
+ Thread.sleep(2000)
+ generateCSVDataFile(spark, idStart = 10, rowNums = 5, csvDataDir)
+ Thread.sleep(5000)
+ thread.interrupt()
+ checkAnswer(
+ sql("select count(*) from streaming.agg_table"),
+ Seq(Row(10)))
+ sql("create datamap p1 on table agg_table using 'preaggregate' as select name, sum(salary) from agg_table group by name")
+ // No data should be loaded into aggregate table as hand-off is not yet fired
+ checkAnswer(sql("select * from agg_table_p1"), Seq())
+ }
+
+ test("test if data is loaded into preaggregate after handoff is fired") {
+ createTable(tableName = "agg_table2", streaming = true, withBatchLoad = false)
+ val identifier = new TableIdentifier("agg_table2", Option("streaming"))
+ val carbonTable = CarbonEnv.getInstance(spark).carbonMetastore.lookupRelation(identifier)(spark)
+ .asInstanceOf[CarbonRelation].metaData.carbonTable
+ val csvDataDir = new File("target/csvdatanew").getCanonicalPath
+ // streaming ingest 10 rows
+ generateCSVDataFile(spark, idStart = 10, rowNums = 5, csvDataDir)
+ val thread = createFileStreamingThread(spark, carbonTable, csvDataDir, intervalSecond = 1,
+ identifier)
+ thread.start()
+ Thread.sleep(2000)
+ generateCSVDataFile(spark, idStart = 10, rowNums = 5, csvDataDir)
+ Thread.sleep(5000)
+ thread.interrupt()
+ checkAnswer(
+ sql("select count(*) from streaming.agg_table2"),
+ Seq(Row(10)))
+ sql("create datamap p1 on table agg_table2 using 'preaggregate' as select name, sum(salary) from agg_table2 group by name")
+ sql("alter table agg_table2 finish streaming")
+ sql("alter table agg_table2 compact 'streaming'")
+ // Data should be loaded into aggregate table as hand-off is fired
+ checkAnswer(sql("select * from agg_table2_p1"),
+ Seq(
+ Row("name_10", 200000.0),
+ Row("name_11", 220000.0),
+ Row("name_12", 240000.0),
+ Row("name_13", 260000.0),
+ Row("name_14", 280000.0)))
+ sql("drop table agg_table2")
+ }
+
+ test("test if data is loaded in aggregate table after handoff is done for streaming table") {
+ createTable(tableName = "agg_table3", streaming = true, withBatchLoad = false)
+ val identifier = new TableIdentifier("agg_table3", Option("streaming"))
+ val carbonTable = CarbonEnv.getInstance(spark).carbonMetastore.lookupRelation(identifier)(spark)
+ .asInstanceOf[CarbonRelation].metaData.carbonTable
+ val csvDataDir = new File("target/csvdatanew").getCanonicalPath
+ // streaming ingest 10 rows
+ generateCSVDataFile(spark, idStart = 10, rowNums = 5, csvDataDir)
+ val thread = createFileStreamingThread(spark, carbonTable, csvDataDir, intervalSecond = 1,
+ identifier)
+ thread.start()
+ Thread.sleep(2000)
+ generateCSVDataFile(spark, idStart = 10, rowNums = 5, csvDataDir)
+ Thread.sleep(5000)
+ thread.interrupt()
+ checkAnswer(
+ sql("select count(*) from streaming.agg_table3"),
+ Seq(Row(10)))
+ sql("alter table agg_table3 finish streaming")
+ sql("alter table agg_table3 compact 'streaming'")
+ sql("create datamap p1 on table agg_table3 using 'preaggregate' as select name, sum(salary) from agg_table3 group by name")
+ // Data should be loaded into aggregate table as hand-off is fired
+ checkAnswer(sql("select * from agg_table3_p1"),
+ Seq(
+ Row("name_10", 200000.0),
+ Row("name_11", 220000.0),
+ Row("name_12", 240000.0),
+ Row("name_13", 260000.0),
+ Row("name_14", 280000.0)))
+ }
+
--- End diff --
Above 3 test cases will have running time + 21 sec of sleep time. I think this is a huge waiting time and will increase the UT running time considerably.
In the complete test file I can find that already we have a sleep time of nearly 30 secs.
@QiangCai ...Can we discuss together and try to find a better solution to avoid this delay in test cases execution
---
Locked
