Intersect multiple secondary index

This optimization tries to intersect multiple secondary indexes if the select conditions introduce them. Previously, as in the IntroduceSelectAccessMethodRule, we would pick the first index to contribute to the access path when there were multiple indexes available. Due to the lack of statistical information, the first one may not be the best choice. Moreover, even we chose the index of the lowest selectivity, it still may not be the best solution. Because we can further reduce the selectivity by intersecting it with the other secondary indexes. Having intersection into the plan will avoid the worst path. Furthermore, if we have statistical information later we can have another option to take into consideration.

Optimization Rule

The logical changes are in the IntroduceSelectAccessMethodRule. After we analyzed the interesting functions and indexes, we pair them up as one to one mapping. (E.g. BTreeAccessMethod -> BTreeIndex On Salary.).

If the primary index appears, we will simply use it as the access path. Because usually the primary index lookup is very rare, if it indeed happens, then it should have a very high chance to be a high selectivity path. Plus, as a clustered index, primary index search is the fastest one. 

If multiple secondary indexes are selected, then we will let each of them to contribute a secondary index to primary index path. A new interface "createSecondaryToPrimaryPlan()" is added to the IAccessMethod for this purpose. (The implementation is required to be functional. Otherwise, different access methods may introduce the conflict states.) Then we use an Intersection logical operator as a join point to intersect the primary keys coming from different secondary index path. 

The following example shows we intersect the BTree,RTree and NgramInvertedIndex before goes to primary index lookup.

drop dataverse test if exists;
create dataverse test;
use dataverse test;

create type tTweet as closed {
id: int32,
location: point,
message: string,
create_at: datetime,
misc: string
}

create dataset dsTweet(tTweet) primary key id;

create index ngram_index on dsTweet(message) type ngram(3);
create index time_index on dsTweet(create_at) type btree;
create index location_index on dsTweet(location) type rtree;

write output to nc1:"rttest/btree-rtree-ngram-intersect.adm";

let $region := create-rectangle(create-point(-128.43007812500002,20.298506037222175), create-point(-64.26992187500002,54.56902589732035))
let $ts_start := datetime("2015-11-11T00:00:00Z")
let $ts_end := datetime("2015-12-18T23:59:59Z")
let $keyword := "hello"
for $t in dataset dsTweet
where $t.create_at >= $ts_start and $t.create_at < $ts_end
and spatial-intersect($t.location, $region)
and contains($t.message, $keyword)
return $t

The corresponding plan is generated as below:

-- DISTRIBUTE_RESULT  |PARTITIONED|
  -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
    -- STREAM_PROJECT  |PARTITIONED|
      -- STREAM_SELECT  |PARTITIONED|
        -- ASSIGN  |PARTITIONED|
          -- STREAM_PROJECT  |PARTITIONED|
            -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
              -- BTREE_SEARCH  |PARTITIONED|
                -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
                  -- INTERSECT  |PARTITIONED|
                    -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
                      -- STABLE_SORT [$$29(ASC)]  |PARTITIONED|
                        -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
                          -- STREAM_PROJECT  |PARTITIONED|
                            -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
                              -- BTREE_SEARCH  |PARTITIONED|
                                -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
                                  -- ASSIGN  |PARTITIONED|
                                    -- EMPTY_TUPLE_SOURCE  |PARTITIONED|
                    -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
                      -- STABLE_SORT [$$31(ASC)]  |PARTITIONED|
                        -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
                          -- STREAM_PROJECT  |PARTITIONED|
                            -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
                              -- LENGTH_PARTITIONED_INVERTED_INDEX_SEARCH  |PARTITIONED|
                                -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
                                  -- ASSIGN  |PARTITIONED|
                                    -- EMPTY_TUPLE_SOURCE  |PARTITIONED|
                    -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
                      -- STABLE_SORT [$$40(ASC)]  |PARTITIONED|
                        -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
                          -- STREAM_PROJECT  |PARTITIONED|
                            -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
                              -- RTREE_SEARCH  |PARTITIONED|
                                -- ONE_TO_ONE_EXCHANGE  |PARTITIONED|
                                  -- ASSIGN  |PARTITIONED|
                                    -- EMPTY_TUPLE_SOURCE  |PARTITIONED|

Hyracks Operator Implementation

Physically, each ETS node will introduce a thread. Thus, the intersection operator must synchronize the upstream input threads in order to generate the correct result. In order to have a pipeline operate, the intersection is implemented as a sort-merge manner. Therefore, each input is required to be sorted. The synchronization is handled by the thread of input zero, which means the thread 0 will call the writer.open/nextFrame/close functions. If we authorize arbitrary threads to do such task, the downstream operator will be confused to synchronize on their locks. The core logical intersection function is as below:

1. do 
   1. find the max input: max
   2. for each input i
      1. keep popping until it matches max. then match++;
      2. if > max, break
   3. If match == inputArity
      1. output max
2. while all inputs haven't been closed.

If any of the input is fully consumed, the operator is closed.

Experiment Evaluation

We did two group of experiments. The first one used a small dataset to test the intersection performance if everything is cached in memory. The second one used a large dataset to test the scenario of loading most of the work from the disk.

Dataset: Two real Twitter datasets. The smaller one is one million records of size 800M, The larger one is ten million record of size 8.2G.

Machine: 4 Cpu, 4G Memory, One disk.

Asterix Instance: 1CC, 2NC, 1 partition per NC.

The AQL query is as following:

use dataverse twitter; 
let $ts_start := datetime("2015-11-23T17:$min_start:00.000Z") 
let $ts_end := datetime("2015-11-23T17:$min_end:03.000Z") 
let $ms_start := date("2010-$month_start-01") 
let $ms_end := date("2010-$month_end-28") 
let $result := for $t in dataset ds_tweets 
               where $t.user.create_at >= $ms_start and $t.user.create_at < $ms_end 
               and $t.create_at >= $ts_start and $t.create_at < $ts_end 
               and $t.place = "Unite State" return $t return count($result)
               return $t
return count($result)

The query selects on the tweet.create_at and the user.create_at.

Each test will run three times:

• 1st. with BTree index on Tweet.create_at only, 
• 2nd. with BTree index on User.create_at only, 
• 3rd. with both indexes presents, consequently, the intersection is introduced.

In Memory case:

Each query will run ten times. We record the time by average the last fives. 

Table 1. Fix the User.create_at $month_start = 01, $month_end = 02, increasing the Tweets.create_at selectivity

Table 2. Fix the Tweet.create_at $min_start = 00, $min_end = 09, increasing the User.create_at selectivity

Table 3. Both Tweet.create_at and User.create_at increasing the selectivity

We can see that intersection is the best one under current settings. The total time reduction is from 5% to 70%. If the two indexes are vary a lot in the selectivity, then the benefit of intersection may not that much. If the two indexes are of the samilar selectivity the intersection can achieve 60% ~ 70% total time reduction.

On disk case

The test dataset is changing to the 8.2G dataset. In order to flush the cache, we load the same dataset to another ds_copy dataset. Every time when we run the selection, we scan this 8.2G ds_copy once to invalidate the cache pages. Due to the slowness of the on disk case, we warm up the query once and record the average time of the next three times.

Review Patch: 

https://asterix-gerrit.ics.uci.edu/#/c/577  and  https://asterix-gerrit.ics.uci.edu/#/c/578