[Spark Load]Got incorrect bucket when distributed by int

[kchen-shanghai]

just not that complex. whatever table that distributed by int type key, loaded by spark load, got incorrect tablet orders.

related codes:
org.apache.doris.load.loadv2.dpp.DppUtils#getHashValue

• The hash value computed without buffer#flip

[kchen-shanghai]

@spaces-X found some bug fixes about spark load submitted by you. could you pls kindly help with this bug ? thx a lot

[wangbo]

Can you supply a user case which may cause this bug?
I want to reproduce it .

[kchen-shanghai]

Can you supply a user case which may cause this bug?
I want to reproduce it .

1. create a table without partition but only distributed like below:
   create table blabla (date date NULL COMMENT "distribute key", total_num bigint(20) SUM NULL COMMENT "sum") ENGINE=OLAP AGGREGATE KEY(date) COMMENT "OLAP" DISTRIBUTED BY HASH(date) BUCKETS 12;
2. use spark load to insert data with date valued '2021-06-07'
3. see the result of below queries:
   select count(1) from blabla where date = '2021-06-07' (got 0, which is not expected)
   select count(1) from blabla where date >= '2021-06-07' (correct)

if we check the plan with "explain select count(1) from blabla where date = '2021-06-07'",
we will see that the scanned tablet is not the same with the one computed by spark load.

[wangbo]

OK， I will verify it

[wangbo]

There are two problems here
1 DppUtils.getHashValue() didn't deal with date type, so no bytes is added here, so result is wrong;
2 I create a table with Integer type and using Spark Load to load data, but I didn't reproduce it; The case is as below;

table:
CREATE TABLE `test_int_bucket` (
  `tinyint_col` tinyint(4) NULL COMMENT "",
  `smallint_col` smallint(6) NULL COMMENT "",
  `int_col` int(11) NULL COMMENT "",
  `bigint_col` bigint(20) NULL COMMENT "",
  `pv_sum` int(11) SUM NULL COMMENT ""
) ENGINE=OLAP
AGGREGATE KEY(`tinyint_col`, `smallint_col`, `int_col`, `bigint_col`)
COMMENT "OLAP"
DISTRIBUTED BY HASH(`tinyint_col`,`smallint_col`,`int_col`,`bigint_col`) BUCKETS 3
PROPERTIES (
"replication_num" = "1",
"in_memory" = "false",
"storage_format" = "DEFAULT"
); 

data:
mysql> select * from test_int_bucket;
+-------------+--------------+---------+------------+--------+
| tinyint_col | smallint_col | int_col | bigint_col | pv_sum |
+-------------+--------------+---------+------------+--------+
|           1 |            1 |       1 |          1 |      1 |
|           4 |            4 |       4 |          4 |      4 |
|           2 |            2 |       2 |          2 |      2 |
|           3 |            3 |       3 |          3 |      3 |
+-------------+--------------+---------+------------+--------+
4 rows in set (0.01 sec)


query:
mysql> select count(1) from test_int_bucket where bigint_col=1;
+----------+
| count(1) |
+----------+
|        1 |
+----------+
1 row in set (0.02 sec)

[kchen-shanghai]

There are two problems here
1 DppUtils.getHashValue() didn't deal with date type, so no bytes is added here, so result is wrong;
2 I create a table with Integer type and using Spark Load to load data, but I didn't reproduce it; The case is as below;

table:
CREATE TABLE `test_int_bucket` (
  `tinyint_col` tinyint(4) NULL COMMENT "",
  `smallint_col` smallint(6) NULL COMMENT "",
  `int_col` int(11) NULL COMMENT "",
  `bigint_col` bigint(20) NULL COMMENT "",
  `pv_sum` int(11) SUM NULL COMMENT ""
) ENGINE=OLAP
AGGREGATE KEY(`tinyint_col`, `smallint_col`, `int_col`, `bigint_col`)
COMMENT "OLAP"
DISTRIBUTED BY HASH(`tinyint_col`,`smallint_col`,`int_col`,`bigint_col`) BUCKETS 3
PROPERTIES (
"replication_num" = "1",
"in_memory" = "false",
"storage_format" = "DEFAULT"
); 

data:
mysql> select * from test_int_bucket;
+-------------+--------------+---------+------------+--------+
| tinyint_col | smallint_col | int_col | bigint_col | pv_sum |
+-------------+--------------+---------+------------+--------+
|           1 |            1 |       1 |          1 |      1 |
|           4 |            4 |       4 |          4 |      4 |
|           2 |            2 |       2 |          2 |      2 |
|           3 |            3 |       3 |          3 |      3 |
+-------------+--------------+---------+------------+--------+
4 rows in set (0.01 sec)


query:
mysql> select count(1) from test_int_bucket where bigint_col=1;
+----------+
| count(1) |
+----------+
|        1 |
+----------+
1 row in set (0.02 sec)

sorry for the wrong case. In the procedure you provided, the distributed key should be only one column like below:
CREATE TABLE test_int_bucket (
int_col int(11) NULL COMMENT "",
pv_sum int(11) SUM NULL COMMENT ""
) ENGINE=OLAP
AGGREGATE KEY(int_col)
COMMENT "OLAP"
DISTRIBUTED BY HASH(int_col) BUCKETS 10
PROPERTIES (
"replication_num" = "1",
"in_memory" = "false",
"storage_format" = "DEFAULT"
);
It is recommended to increase the bucket number so that it would be easier to to reproduce it.

Another option is to change the query to below:
select count(1) from test_int_bucket where tinyint_col=1 and smallint_col=1 and int_col=1 and bigint_col=1;
The key is to make the query plan scan ONLY ONE BUCKET that contains the target data. The problem here is that the query goes to the wrong tablet, not that the data itself (i.e. if scanning all tablet, the result wont go wrong).

[kchen-shanghai]

if checking 'explain select count(1) from test_int_bucket where bigint_col=1;' on the provided case, the query plan should be a full db scan on all buckets. In order to reproduce this issue the qe must be a scan on only one bucket that computed by FE.
As the reference code here in org.apache.doris.analysis.IntLiteral#getHashValue, we can see the calling of 'buffer.flip();' which is different with spark load (which does not flip). @wangbo

[wangbo]

I have reproduce it in my in environment and calling buffer.flip can fix it.
Thanks for reporting a bug.
Could you pull a request for this case?