Presto+CarbonData optimization work discussion

[Liang Chen]

Hi

Below are some proposed items for Presto optimization:

1)  Remove the extra loops for data conversion in Presto Format to increase
the performance.
2) Modularize and optimize the filters .
3) Optimize the Carbondata Metadata reading.
4) Lazy decoding of the dictionary.
5) Batch reading of the rows in bulk to get good performance.
6) Vectorized reader implementation to read the columns instead of rows.

Please discuss and comment.

Regards
Liang

[Liang Chen]

Hi

For -- 4) Lazy decoding of the dictionary,  just i tested 180 millions rows data with the script:
"select province,sum(age),count(*) from presto_carbondata group by province order by province"

Spark integration module has "dictionary lazy decode", presto doesn't have "dictionary lazy decode", the performance is 4.5 times difference, so "dictionary lazy decode" might much help to improve aggregation performance.

The detail test result as below :

1. Presto+CarbonData is 9 second:
presto:default> select province,sum(age),count(*) from presto_carbondata group by province order by province;
 province |  _col1   |  _col2
----------+----------+---------
 AB       | 57442740 | 1385010
 BC       | 57488826 | 1385580
 MB       | 57564702 | 1386510
 NB       | 57599520 | 1386960
 NL       | 57446592 | 1383774
 NS       | 57448734 | 1384272
 NT       | 57534228 | 1386936
 NU       | 57506844 | 1385346
 ON       | 57484956 | 1384470
 PE       | 57325164 | 1379802
 QC       | 57467886 | 1385076
 SK       | 57385152 | 1382364
 YT       | 57377556 | 1383900
(13 rows)

Query 20170720_022833_00004_c9ky2, FINISHED, 1 node
Splits: 55 total, 55 done (100.00%)
0:09 [18M rows, 34.3MB] [1.92M rows/s, 3.65MB/s]

2.Spark+CarbonData is :2 seconds
scala> benchmark { carbon.sql("select province,sum(age),count(*) from presto_carbondata group by province order by province").show }
+--------+--------+--------+
|province|sum(age)|count(1)|
+--------+--------+--------+
|      AB|57442740| 1385010|
|      BC|57488826| 1385580|
|      MB|57564702| 1386510|
|      NB|57599520| 1386960|
|      NL|57446592| 1383774|
|      NS|57448734| 1384272|
|      NT|57534228| 1386936|
|      NU|57506844| 1385346|
|      ON|57484956| 1384470|
|      PE|57325164| 1379802|
|      QC|57467886| 1385076|
|      SK|57385152| 1382364|
|      YT|57377556| 1383900|
+--------+--------+--------+

2109.346231ms

[ravipesala]

Hi Liang,

I see that province column data is not big, so I guess it hardly make any
impact with lazy decoding in this scenario. Can you do one more test by
excluding the province from dictionary in both presto and spark
integrations. It will tell whether it is really a lazy decoding issue or
not.

Regards,
Ravindra

On 20 July 2017 at 08:04, Liang Chen <[hidden email]> wrote:

> Hi
>
> For -- 4) Lazy decoding of the dictionary,  just i tested 180 millions rows
> data with the script:
> "select province,sum(age),count(*) from presto_carbondata group by province
> order by province"
>
> Spark integration module has "dictionary lazy decode", presto doesn't have
> "dictionary lazy decode", the performance is 4.5 times difference, so
> "dictionary lazy decode" might much help to improve aggregation
> performance.
>
> The detail test result as below :
>
> *1. Presto+CarbonData is 9 second:*
> presto:default> select province,sum(age),count(*) from presto_carbondata
> group by province order by province;
>  province |  _col1   |  _col2
> ----------+----------+---------
>  AB       | 57442740 | 1385010
>  BC       | 57488826 | 1385580
>  MB       | 57564702 | 1386510
>  NB       | 57599520 | 1386960
>  NL       | 57446592 | 1383774
>  NS       | 57448734 | 1384272
>  NT       | 57534228 | 1386936
>  NU       | 57506844 | 1385346
>  ON       | 57484956 | 1384470
>  PE       | 57325164 | 1379802
>  QC       | 57467886 | 1385076
>  SK       | 57385152 | 1382364
>  YT       | 57377556 | 1383900
> (13 rows)
>
> Query 20170720_022833_00004_c9ky2, FINISHED, 1 node
> Splits: 55 total, 55 done (100.00%)
> 0:09 [18M rows, 34.3MB] [1.92M rows/s, 3.65MB/s]
>
> *2.Spark+CarbonData is :2 seconds*
> scala> benchmark { carbon.sql("select province,sum(age),count(*) from
> presto_carbondata group by province order by province").show }
> +--------+--------+--------+
> |province|sum(age)|count(1)|
> +--------+--------+--------+
> |      AB|57442740| 1385010|
> |      BC|57488826| 1385580|
> |      MB|57564702| 1386510|
> |      NB|57599520| 1386960|
> |      NL|57446592| 1383774|
> |      NS|57448734| 1384272|
> |      NT|57534228| 1386936|
> |      NU|57506844| 1385346|
> |      ON|57484956| 1384470|
> |      PE|57325164| 1379802|
> |      QC|57467886| 1385076|
> |      SK|57385152| 1382364|
> |      YT|57377556| 1383900|
> +--------+--------+--------+
>
> 2109.346231ms
>
>
>
> --
> View this message in context: http://apache-carbondata-dev-
> mailing-list-archive.1130556.n5.nabble.com/Presto-
> CarbonData-optimization-work-discussion-tp18509p18522.html
> Sent from the Apache CarbonData Dev Mailing List archive mailing list
> archive at Nabble.com.
>

--
Thanks & Regards,
Ravi

[Liang Chen-2]

Hi Ravi

Thanks for your comment.

I tested again with excluding province as dictionary.  In spark, the query
time is around 3 seconds, in presto same is 9 seconds.  so for this query
case(short string), dictionary lazy decode might not be the key factor.

Regards
Liang

2017-07-20 10:56 GMT+08:00 Ravindra Pesala <[hidden email]>:

> Hi Liang,
>
> I see that province column data is not big, so I guess it hardly make any
> impact with lazy decoding in this scenario. Can you do one more test by
> excluding the province from dictionary in both presto and spark
> integrations. It will tell whether it is really a lazy decoding issue or
> not.
>
> Regards,
> Ravindra
>
> On 20 July 2017 at 08:04, Liang Chen <[hidden email]> wrote:
>
> > Hi
> >
> > For -- 4) Lazy decoding of the dictionary,  just i tested 180 millions
> rows
> > data with the script:
> > "select province,sum(age),count(*) from presto_carbondata group by
> province
> > order by province"
> >
> > Spark integration module has "dictionary lazy decode", presto doesn't
> have
> > "dictionary lazy decode", the performance is 4.5 times difference, so
> > "dictionary lazy decode" might much help to improve aggregation
> > performance.
> >
> > The detail test result as below :
> >
> > *1. Presto+CarbonData is 9 second:*
> > presto:default> select province,sum(age),count(*) from presto_carbondata
> > group by province order by province;
> >  province |  _col1   |  _col2
> > ----------+----------+---------
> >  AB       | 57442740 | 1385010
> >  BC       | 57488826 | 1385580
> >  MB       | 57564702 | 1386510
> >  NB       | 57599520 | 1386960
> >  NL       | 57446592 | 1383774
> >  NS       | 57448734 | 1384272
> >  NT       | 57534228 | 1386936
> >  NU       | 57506844 | 1385346
> >  ON       | 57484956 | 1384470
> >  PE       | 57325164 | 1379802
> >  QC       | 57467886 | 1385076
> >  SK       | 57385152 | 1382364
> >  YT       | 57377556 | 1383900
> > (13 rows)
> >
> > Query 20170720_022833_00004_c9ky2, FINISHED, 1 node
> > Splits: 55 total, 55 done (100.00%)
> > 0:09 [18M rows, 34.3MB] [1.92M rows/s, 3.65MB/s]
> >
> > *2.Spark+CarbonData is :2 seconds*
> > scala> benchmark { carbon.sql("select province,sum(age),count(*) from
> > presto_carbondata group by province order by province").show }
> > +--------+--------+--------+
> > |province|sum(age)|count(1)|
> > +--------+--------+--------+
> > |      AB|57442740| 1385010|
> > |      BC|57488826| 1385580|
> > |      MB|57564702| 1386510|
> > |      NB|57599520| 1386960|
> > |      NL|57446592| 1383774|
> > |      NS|57448734| 1384272|
> > |      NT|57534228| 1386936|
> > |      NU|57506844| 1385346|
> > |      ON|57484956| 1384470|
> > |      PE|57325164| 1379802|
> > |      QC|57467886| 1385076|
> > |      SK|57385152| 1382364|
> > |      YT|57377556| 1383900|
> > +--------+--------+--------+
> >
> > 2109.346231ms
> >
> >
> >
> > --
> > View this message in context: http://apache-carbondata-dev-
> > mailing-list-archive.1130556.n5.nabble.com/Presto-
> > CarbonData-optimization-work-discussion-tp18509p18522.html
> > Sent from the Apache CarbonData Dev Mailing List archive mailing list
> > archive at Nabble.com.
> >
>
>
>
> --
> Thanks & Regards,
> Ravi
>

[rui qin]

For -- 6) spark has the vectorized feature,but not in presto.How to implement it？

[rui qin]

For presto carbon data connector,I have two suggestions：
1) Creating a carbon data table though spark-shell,spark-shell can store the table metadata into the hive metastore. Currently, the loading metadata is from the file and then cached in presto carbon connector.
This creates a problem that I have deleted a table and recreated the same table,which is a phenomenon of metadata inconsistency by presto query again.The only solution is to restart the presto cluster.In order to avoid it,I want to load the carbon data table metadata from the hive metastore.
2) For the  SegmentTaskIndexStore class,its's segmentProperties can cache the table segment detail.But if I have deleted a table, the 'segmentProperties' always holds the table segment detail,until restarting presto.It may be cause memory leak.I don't suggest to cache segment detail.The lruCache of SegmentTaskIndexStore don't cause the above phenomenon.

[bhavya411]

I have created a pull request 1190  for Presto Optimization where we have done following changes to improve the performance

1. Removed unnecessary loops from the integration code to make it more efficient.

2. Implemented Lazy Blocks as is being used in case of ORC.

3. Improved dictionary decoding to have better results.

I have run this on my local machine for 2 GB data and results are attached with this email, we see an improvement in almost all TPCH queries that we have run.

Thanks and regards

Bhavya

On Thu, Jul 20, 2017 at 12:21 PM, rui qin <[hidden email]> wrote:

For -- 6) spark has the vectorized feature,but not in presto.How to implement
it？



--
View this message in context: http://apache-carbondata-dev-mailing-list-archive.1130556.n5.nabble.com/Presto-CarbonData-optimization-work-discussion-tp18509p18548.html

Sent from the Apache CarbonData Dev Mailing List archive mailing list archive at Nabble.com.

[rui qin]

Please provide the statement to create the table in carbon data.And the block size is 1024MB?

[bhavya411]

Please find the queries below , yes we are using default 1024 GB as block size

 CREATE TABLE NATION ( N_NATIONKEY INT, N_NAME STRING, N_REGIONKEY INT, N_COMMENT STRING ) STORED BY 'carbondata';

CREATE TABLE REGION ( R_REGIONKEY INT , R_NAME STRING , R_COMMENT STRING) STORED BY 'carbondata';

CREATE TABLE PART ( P_PARTKEY INT , P_NAME STRING ,P_MFGR STRING ,P_BRAND STRING , P_TYPE STRING ,P_SIZE INT ,P_CONTAINER STRING , P_RETAILPRICE DECIMAL(15,2) ,P_COMMENT STRING ) STORED BY 'carbondata';

CREATE TABLE SUPPLIER ( S_SUPPKEY INT ,S_NAME STRING ,S_ADDRESS STRING ,S_NATIONKEY INT ,S_PHONE STRING , S_ACCTBAL DECIMAL(15,2) , S_COMMENT STRING ) STORED BY 'carbondata';

CREATE TABLE PARTSUPP ( PS_PARTKEY INT ,PS_SUPPKEY INT ,PS_AVAILQTY INT , PS_SUPPLYCOST DECIMAL(15,2) ,PS_COMMENT STRING ) STORED BY 'carbondata';


CREATE TABLE CUSTOMER (C_CUSTKEY INT ,C_NAME STRING ,C_ADDRESS STRING ,C_NATIONKEY INT ,C_PHONE STRING ,C_ACCTBAL DECIMAL(15,2) ,C_MKTSEGMENT STRING ,C_COMMENT STRING) STORED BY 'carbondata';

CREATE TABLE ORDERS ( O_ORDERKEY INT ,O_CUSTKEY INT ,O_ORDERSTATUS STRING ,O_TOTALPRICE DECIMAL(15,2) ,O_ORDERDATE DATE ,O_ORDERPRIORITY STRING,O_CLERK STRING ,O_SHIPPRIORITY INT ,O_COMMENT STRING ) STORED BY 'carbondata';

CREATE TABLE LINEITEM (L_ORDERKEY INT ,L_PARTKEY INT ,L_SUPPKEY INT ,L_LINENUMBER INT ,L_QUANTITY DECIMAL(15,2) ,L_EXTENDEDPRICE DECIMAL(15,2) ,L_DISCOUNT DECIMAL(15,2) ,L_TAX DECIMAL(15,2) ,L_RETURNFLAG STRING ,L_LINESTATUS STRING ,L_SHIPDATE DATE ,L_COMMITDATE DATE ,L_RECEIPTDATE DATE ,L_SHIPINSTRUCT STRING ,L_SHIPMODE STRING ,L_COMMENT STRING ) STORED BY 'carbondata';

[rui qin]

Hi,
   Based on tpch test, I have run the 10 GB data and results are attached with this email.We see a little improvement. However,with the increase in concurrency, the sql execution time will drop.Compared with the previous test, it has not been improved.The block size is 1024 MB and we use your create statement.





prestoTest.xlsx

[bhavya411]

I am already working on concurrency level improvement as well, this was
done for single query execution as we are trying to make single queries
execute much faster. As we can see in the attached results there is an
improvement in almost all the queries so the expectation was that in
concurrent queries also it will improve. Are you running the same queries
concurrently or different queries and can you also tell me how you are
running these queries concurrently.

Regards
Bhavya

On Fri, Jul 28, 2017 at 1:29 PM, rui qin <[hidden email]> wrote:

> Hi,
>    Based on tpch test, I have run the 10 GB data and results are attached
> with this email.We see a little improvement. However,with the increase in
> concurrency, the sql execution time will drop.Compared with the previous
> test, it has not been improved.The block size is 1024 MB and we use your
> create statement.
>
>
>
>
>
> prestoTest.xlsx
> <http://apache-carbondata-dev-mailing-list-archive.1130556.
> n5.nabble.com/file/n19002/prestoTest.xlsx>
>
>
>
> --
> View this message in context: http://apache-carbondata-dev-
> mailing-list-archive.1130556.n5.nabble.com/Presto-
> CarbonData-optimization-work-discussion-tp18509p19002.html
> Sent from the Apache CarbonData Dev Mailing List archive mailing list
> archive at Nabble.com.
>

[Liang Chen]

Hi

Based on pull request 1307, the latest test result as below, the performance
be improved 3 times.

presto:default> select province,sum(age),count(*) from presto_carbon_dict
group by province order by province;
 province |  _col1   |  _col2
----------+----------+---------
 AB       | 57442740 | 1385010
 BC       | 57488826 | 1385580
 MB       | 57564702 | 1386510
 NB       | 57599520 | 1386960
 NL       | 57446592 | 1383774
 NS       | 57448734 | 1384272
 NT       | 57534228 | 1386936
 NU       | 57506844 | 1385346
 ON       | 57484956 | 1384470
 PE       | 57325164 | 1379802
 QC       | 57467886 | 1385076
 SK       | 57385152 | 1382364
 YT       | 57377556 | 1383900
(13 rows)

Query 20170902_033821_00006_h6g24, FINISHED, 1 node
Splits: 50 total, 50 done (100.00%)
0:03 [18M rows, 0B] [6.62M rows/s, 0B/s]


Regards
Liang


Liang Chen wrote

> Hi
>
> For -- 4) Lazy decoding of the dictionary,  just i tested 180 millions
> rows data with the script:
> "select province,sum(age),count(*) from presto_carbondata group by
> province order by province"
>
> Spark integration module has "dictionary lazy decode", presto doesn't have
> "dictionary lazy decode", the performance is 4.5 times difference, so
> "dictionary lazy decode" might much help to improve aggregation
> performance.
>
> The detail test result as below :

*
> 1. Presto+CarbonData is 9 second:
*

> presto:default> select province,sum(age),count(*) from presto_carbondata
> group by province order by province;
>  province |  _col1   |  _col2
> ----------+----------+---------
>  AB       | 57442740 | 1385010
>  BC       | 57488826 | 1385580
>  MB       | 57564702 | 1386510
>  NB       | 57599520 | 1386960
>  NL       | 57446592 | 1383774
>  NS       | 57448734 | 1384272
>  NT       | 57534228 | 1386936
>  NU       | 57506844 | 1385346
>  ON       | 57484956 | 1384470
>  PE       | 57325164 | 1379802
>  QC       | 57467886 | 1385076
>  SK       | 57385152 | 1382364
>  YT       | 57377556 | 1383900
> (13 rows)
>
> Query 20170720_022833_00004_c9ky2, FINISHED, 1 node
> Splits: 55 total, 55 done (100.00%)
> 0:09 [18M rows, 34.3MB] [1.92M rows/s, 3.65MB/s]

*
> 2.Spark+CarbonData is :2 seconds
*

> scala> benchmark { carbon.sql("select province,sum(age),count(*) from
> presto_carbondata group by province order by province").show }
> +--------+--------+--------+
> |province|sum(age)|count(1)|
> +--------+--------+--------+
> |      AB|57442740| 1385010|
> |      BC|57488826| 1385580|
> |      MB|57564702| 1386510|
> |      NB|57599520| 1386960|
> |      NL|57446592| 1383774|
> |      NS|57448734| 1384272|
> |      NT|57534228| 1386936|
> |      NU|57506844| 1385346|
> |      ON|57484956| 1384470|
> |      PE|57325164| 1379802|
> |      QC|57467886| 1385076|
> |      SK|57385152| 1382364|
> |      YT|57377556| 1383900|
> +--------+--------+--------+
>
> 2109.346231ms

--
Sent from: http://apache-carbondata-dev-mailing-list-archive.1130556.n5.nabble.com/