[DISCUSSION] Distributed Index Cache Server

[kunalkapoor]

Hi All,

Carbon currently caches all block/blocklet datamap index information into
the driver. And for bloom type of datamap, it can prune the splits in a
distributed way using distributed datamap pruning. In the first case, there
are limitations like driver memory scale up and reusing of one driver cache
by others is not possible. In the second case, there are limitations like
there is no guarantee that the next query goes to the same executor to
reuse the cache.


Based on the above problems there is a need to have a centralised index
cache server.


Please find below the link for the design document.


https://docs.google.com/document/d/161NXxrKLPucIExkWip5mX00x2iOPH6bvsuQnCzzp47E/edit?ts=5c542ab4#heading=h.x0qaehgkncz5



Thanks

Kunal Kapoor

[kunalkapoor]

+ JIRA link for tracking purpose


On Tue, Feb 5, 2019 at 4:27 PM Kunal Kapoor <[hidden email]>
wrote:

> Hi All,
>
> Carbon currently caches all block/blocklet datamap index information into
> the driver. And for bloom type of datamap, it can prune the splits in a
> distributed way using distributed datamap pruning. In the first case, there
> are limitations like driver memory scale up and reusing of one driver cache
> by others is not possible. In the second case, there are limitations like
> there is no guarantee that the next query goes to the same executor to
> reuse the cache.
>
>
> Based on the above problems there is a need to have a centralised index
> cache server.
>
>
> Please find below the link for the design document.
>
>
>
> https://docs.google.com/document/d/161NXxrKLPucIExkWip5mX00x2iOPH6bvsuQnCzzp47E/edit?ts=5c542ab4#heading=h.x0qaehgkncz5
>
>
>
> Thanks
>
> Kunal Kapoor
>
>
>

[kunalkapoor]

+ JIRA link for tracking purpose

https://issues.apache.org/jira/browse/CARBONDATA-3288

On Tue, Feb 5, 2019 at 4:36 PM Kunal Kapoor <[hidden email]>
wrote:

> + JIRA link for tracking purpose
>
>
> On Tue, Feb 5, 2019 at 4:27 PM Kunal Kapoor <[hidden email]>
> wrote:
>
>> Hi All,
>>
>> Carbon currently caches all block/blocklet datamap index information into
>> the driver. And for bloom type of datamap, it can prune the splits in a
>> distributed way using distributed datamap pruning. In the first case, there
>> are limitations like driver memory scale up and reusing of one driver cache
>> by others is not possible. In the second case, there are limitations like
>> there is no guarantee that the next query goes to the same executor to
>> reuse the cache.
>>
>>
>> Based on the above problems there is a need to have a centralised index
>> cache server.
>>
>>
>> Please find below the link for the design document.
>>
>>
>>
>> https://docs.google.com/document/d/161NXxrKLPucIExkWip5mX00x2iOPH6bvsuQnCzzp47E/edit?ts=5c542ab4#heading=h.x0qaehgkncz5
>>
>>
>>
>> Thanks
>>
>> Kunal Kapoor
>>
>>
>>

[litao]

I have some problem about index server driver.
1) How does driver know excutor’s data distribution? How does the driver
choice the executor to work?
2) From the design document it seems that under the index cache server index
data store under each executor, there is no need for Use of Cache Layer.
3) Under carbon driver, if index cache server failure we need to reuse
driver side cache. If we can abstract driver side cache into a cache layer?




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

[xuchuanyin]

Hi kunal, can you attach the document directly to the jira? I cannot access
the doc on google drive. Thanks.



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

[litao]

Hi, guys
I have some problem about index server driver.
1) How does driver know excutor’s data distribution? How does the driver choice the executor to
work?
2) From the design document it seems that under the index cache server index data store under each executor, there is no need for Use of Cache
Layer.
3) Under carbon driver, if index cache server failure we need to reuse driver side cache. If we can abstract driver side cache into a cache layer?



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

[litao]

some discussions：
1. The first time it has to load all the datamaps based on the list of
segments provided by the main server. Index pruning will happen from the
cached datamaps.
        -> Does the main server means the carbon driver?
        Kunal : Yes main server means carbon driver.. I will update the same
in the design.
2. The pruned splits can either be written to a file if the number of splits
are too many or can be sent back to the carbon driver directly.
        -> Is carbon driver has two ways to get index. One is directly
return, another is by load file from hdfs that the path is return to carbon
driver?
        Kunal : Correct, if the splits count is huge like 1 million then
serialization cost would be too huge. Therefore it is better to write to a
file so that the carbon driver can read the files and generate splits
directly.
3. If the number of splits increases the threshold then, the index driver
will create a multi block split on the directory in which the files are
written and serialize them to the carbon driver.
 What is the multi block split mean?
Kunal : Multi-Block split means that it contains more than 1 split. This is
a way to tell the carbon driver that the splits are written to a file with
the location present in Multi-block split object and the driver can directly
read from there.
4. The server should be called when the size of the index files of the table
is more than 1GB. If the table size is less than the main driver can prune
        -> How can carbon driver decide to use index server or carbon driver
cache?
        Kunal : The carbon driver will read the table status file and
generate the LoadMetaDataDetails, Index size stored in the load details can
be used to calculate the total index size for the table. If more than the
configured value then the index server can be called otherwise the carbon
driver can prune the splits(how current pruning is done).
5. The index files to be divided between the executors should be based on
size and not count.
        -> Does consider whether the index of the same table is distributed
on the same excutor?
        Kunal : Yes, it will consider because the index driver will have a
mapping for the same. The idea here is that the size of datamaps that are
handled by the executors have to be same.
6. Dynamic allocation for index server should be false
        -> What does this mean?
        Kunal : Dynamic executor allocation, this property is exposed by
spark where the executor processes are killed once their jobs are complete
if set to true. This is not good for the server as then it will have to
reload the datamaps.
7. Fallback
        -> If index server recover, carbon driver need to reuse the index
server cache instead of self lru cache.
        Kunal : Correct, will update the same in the design





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

[kunalkapoor]

Hi Xuchuanyin,
Uploaded the design in PDF format to the JIRA.

Thanks
Kunal Kapoor

On Tue, Feb 12, 2019 at 1:58 PM xuchuanyin <[hidden email]> wrote:

> Hi kunal, can you attach the document directly to the jira? I cannot access
> the doc on google drive. Thanks.
>
>
>
> --
> Sent from:
> http://apache-carbondata-dev-mailing-list-archive.1130556.n5.nabble.com/
>

[manishgupta88]

+1

1. Add the impacted areas in design document.
2. If any executor goes down then update the index cache to executor mapping
in driver accordingly.
3. Even though the cache would be divided based on index files, the minimum
unit of cache need to be fixed. Example: 1 segment cache should belong to 1
executor only.
4. One possible suggestion: Instead of reading the splits in Carbon driver,
let each executor in index server write to a file and pass the
List[FilePaths] to the driver and let each Carbon Executors to read from
that path. This is the case when the number of splits exceed the threshold.

Regards
Manish Gupta



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

[xuchuanyin]

Hi Kunal,
  IndexServer is quiet an efficient method to solve the problem of index
cache and it's great that someone finally tries to implement this. However
after I went through your design document, I get some questions for this and
I'll explain those as following:

1. For the 'backgroud' chapter, I think actually it is the type of pruning
(distribute-pruning or not) that matters, not the type of datamaps (default
or bloomfilter).

2. Extensibility of the IndexServer
2.1 In the design document, why do you finally choose 'one more spark
cluster' as the IndexServer?

2.2 Have you considered other types of IndexServer such as a DB, another
in-memory storage engine or even treat the current implementation as an
embedded IndexServer? If yes, Will the base IndexServer be enough extensible
to support other them during your implementation and design?

2.3 What are the interfaces that the IndexServer will expose to offer
service? I also didn't get this info.

3. For the IndexServer, will multiple tenants also be OK?

4. During coding, will IndexServer be in a separate module?




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

[kunalkapoor]

Hi Manish,
Thank you for the suggestions.

1. I will add the impacted areas to the design document.
2. Yes the mapping has to be updated when a executor is down and when it
get back up the scheduling of the splits has to be done accordingly. Same
will be updated in the design.
3. I think the distribution should be based on the index files and not the
segments, so that when the user has set only 1 segment for the query even
then distribution will happen.
4. Already updated in the design

On Tue, Feb 12, 2019 at 10:18 PM manishgupta88 <[hidden email]>
wrote:

> +1
>
> 1. Add the impacted areas in design document.
> 2. If any executor goes down then update the index cache to executor
> mapping
> in driver accordingly.
> 3. Even though the cache would be divided based on index files, the minimum
> unit of cache need to be fixed. Example: 1 segment cache should belong to 1
> executor only.
> 4. One possible suggestion: Instead of reading the splits in Carbon driver,
> let each executor in index server write to a file and pass the
> List[FilePaths] to the driver and let each Carbon Executors to read from
> that path. This is the case when the number of splits exceed the threshold.
>
> Regards
> Manish Gupta
>
>
>
> --
> Sent from:
> http://apache-carbondata-dev-mailing-list-archive.1130556.n5.nabble.com/
>

[kunalkapoor]

Hi Xuchuanyin,
Thank you for the suggestion/questions.

1. You are right the only thing in the spotlight is the pruning, the
datamaps are not important because we would support all types of datamaps.
The bloom datamap line was just an example to illustrate that for bloom we
are already using distributed datamap pruning. I will re-write the same in
a better way.

2.1 We want the index server to run in a different cluster so that it is
centralised.

2.2 We had considered the possibility of using an in-memory DB but the same
problems will happen with huge split load(1 million or more). Also other
solutions like Elasticsearch which would be much faster but the
implementation would have to be done from scratch. For now we are starting
the requirement with a less error prone method because the existing pruning
logic has to be moved from driver to executor. No new logic is being
introduced. But we can surely integrate other solutions in the future.

2.3 The start and stop of index server/client is the only new interface
that will be provided, rest all the existing interfaces will be reused. Ill
update the same in the design soon.

3. Yes Index server will support multi-tenant, we are currently trying to
figure out the best way to authorise and authenticate the access for
multiple users.

4. Yes a seperate module would be create but just to start the server and
client. The other logic would not be moved to this module.

Thanks
Kunal Kapoor




On Wed, Feb 13, 2019 at 6:59 AM xuchuanyin <[hidden email]> wrote:

> Hi Kunal,
>   IndexServer is quiet an efficient method to solve the problem of index
> cache and it's great that someone finally tries to implement this. However
> after I went through your design document, I get some questions for this
> and
> I'll explain those as following:
>
> 1. For the 'backgroud' chapter, I think actually it is the type of pruning
> (distribute-pruning or not) that matters, not the type of datamaps (default
> or bloomfilter).
>
> 2. Extensibility of the IndexServer
> 2.1 In the design document, why do you finally choose 'one more spark
> cluster' as the IndexServer?
>
> 2.2 Have you considered other types of IndexServer such as a DB, another
> in-memory storage engine or even treat the current implementation as an
> embedded IndexServer? If yes, Will the base IndexServer be enough
> extensible
> to support other them during your implementation and design?
>
> 2.3 What are the interfaces that the IndexServer will expose to offer
> service? I also didn't get this info.
>
> 3. For the IndexServer, will multiple tenants also be OK?
>
> 4. During coding, will IndexServer be in a separate module?
>
>
>
>
> --
> Sent from:
> http://apache-carbondata-dev-mailing-list-archive.1130556.n5.nabble.com/
>

[xuchuanyin]

Hi kunal, I think we can go further for 2.3 & 4.

For 4, I think all functions of IndexServer should be in an individual module. We can think of the IndexServer as an enhancement component for Carbondata. And inside that module we handle the actual pruning logic. On the other side, if we do not have this component, there will be no pruning at all.

As a consequence, for 2.3, I think the IndexServer should provide interfaces that will provide pruning services. For example it accepts expressions and returns pruning result.

I think only in this way can the IndexServer be more extensible to meet higher requirements.

[dhatchayani]

Hi Kunal,

This feature looks great from the design.

Still, I need some more clarifications on the below points.

(1) How segment deletion will be handled? Whether the next query takes care
of clearing this segment and update the driver map or the delete operation
will update?
(2) Is there any impact on the CLEAN FILES command?
(3) Is there any impact on the COLUMN_META_CACHE property? This is a session
property and can be changed through ALTER command. If this property is
altered, accordingly the current cache implementation will invalidate the
datamap cache, in required cases.
(4) Executor shut down/failures will be handled by the spark cluster
manager? In between query execution, if some executor fails, then the tasks
will be re-launched in any other available executors?



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

[yaojinguo]

nice feature. I still have some questions:
1. what's the  impact on set carbon.input.segments command? Index Cache
Server may make the query slower.
2. what's your plan of this feature.



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

[kunalkapoor]

Hi Dhatchayani,
1. The next query will take care of removing the cache for the deleted
segments. The request is designed to contain the invalid segments as well,
so that the corresponding datamaps can be removed from the cache.
2. No impact on clean files command.
3. ColumnCache will behave in the same way. If alter command is fired then
the cache would be changed accordingly.
4. This is a valid point, the query retry configuration should be disabled
so that the datamaps are caches in the assigned executor only. Even if the
query fails then the carbon driver will take care of the pruning.

Thanks
Kunal Kapoor

On Thu, Feb 14, 2019 at 11:56 PM dhatchayani <[hidden email]>
wrote:

> Hi Kunal,
>
> This feature looks great from the design.
>
> Still, I need some more clarifications on the below points.
>
> (1) How segment deletion will be handled? Whether the next query takes care
> of clearing this segment and update the driver map or the delete operation
> will update?
> (2) Is there any impact on the CLEAN FILES command?
> (3) Is there any impact on the COLUMN_META_CACHE property? This is a
> session
> property and can be changed through ALTER command. If this property is
> altered, accordingly the current cache implementation will invalidate the
> datamap cache, in required cases.
> (4) Executor shut down/failures will be handled by the spark cluster
> manager? In between query execution, if some executor fails, then the tasks
> will be re-launched in any other available executors?
>
>
>
> --
> Sent from:
> http://apache-carbondata-dev-mailing-list-archive.1130556.n5.nabble.com/
>

[kunalkapoor]

Hi yaojinguo,
1. The  carbon.input.segments property will restrict the segments that have
to be accessed. Therefore the carbon driver will send on the specified
segments for pruning.
Why do you think that the Cache Server will slow down the query?

On Fri, Feb 15, 2019 at 2:52 PM yaojinguo <[hidden email]> wrote:

> nice feature. I still have some questions:
> 1. what's the  impact on set carbon.input.segments command? Index Cache
> Server may make the query slower.
> 2. what's your plan of this feature.
>
>
>
> --
> Sent from:
> http://apache-carbondata-dev-mailing-list-archive.1130556.n5.nabble.com/
>

[kunalkapoor]

Hi xuchuanyin,
I will expose an interface and the put the same in the design document soon.

Thanks for the feedback
Kunal Kapoor


On Wed, Feb 13, 2019 at 8:04 PM ChuanYin Xu <[hidden email]> wrote:

> Hi kunal, I think we can go further for 2.3 & 4.
>
> For 4, I think all functions of IndexServer should be in an individual
> module. We can think of the IndexServer as an enhancement component for
> Carbondata. And inside that module we handle the actual pruning logic. On
> the other side, if we do not have this component, there will be no pruning
> at all.
>
> As a consequence, for 2.3, I think the IndexServer should provide
> interfaces that will provide pruning services. For example it accepts
> expressions and returns pruning result.
>
> I think only in this way can the IndexServer be more extensible to meet
> higher requirements.
>
>

[kunalkapoor]

Hi xuchuanyin,
I have uploaded the version2 of the design document with the desired
changes.
Please review and let me know if anything is missing or needs to be changed.

Thanks
Kunal Kapoor

On Mon, Feb 18, 2019 at 12:15 PM Kunal Kapoor <[hidden email]>
wrote:

> Hi xuchuanyin,
> I will expose an interface and the put the same in the design document
> soon.
>
> Thanks for the feedback
> Kunal Kapoor
>
>
> On Wed, Feb 13, 2019 at 8:04 PM ChuanYin Xu <[hidden email]>
> wrote:
>
>> Hi kunal, I think we can go further for 2.3 & 4.
>>
>> For 4, I think all functions of IndexServer should be in an individual
>> module. We can think of the IndexServer as an enhancement component for
>> Carbondata. And inside that module we handle the actual pruning logic. On
>> the other side, if we do not have this component, there will be no pruning
>> at all.
>>
>> As a consequence, for 2.3, I think the IndexServer should provide
>> interfaces that will provide pruning services. For example it accepts
>> expressions and returns pruning result.
>>
>> I think only in this way can the IndexServer be more extensible to meet
>> higher requirements.
>>
>>

[xuchuanyin]

Hi kunal,

At last I'd suggest again that the code for pruning procedure should be
moved to a separate module.
The earlier we do this, the easier will be if we want to implement other
types of IndexServer later.



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