[Proposal] The 2nd version of implementation for materialised view

[zhangxinyu1]

The motivation of 'materialised view' is in issue #5218 .
The implementation of 'Materialised View' is diveded into two parts. One is the management of DerivedDatasource, and the other is optimizing query with DerivedDatasource.

Management of DerivedDatasource

Each derived-datasource is managed by a derived-datasource supervisor(like kafkaSupervisor).

CREATE

DerivedDatasource is created when user submits a derived-datasource supervisor. The json file of derived-datasource supervisor should include base datasource name, dimensions and metrics at least. DerivedDatasource name can be setted by users or generated by base datasource name, dimensions and metrics. Segment granularity and query granularity should keep the same with base datasource. The information is stored as metadata of supervisors.

MAINTAIN

A derived-datasource supervisor should make sure timeline and segment versions are the same with its base datasource (It can sovle the problem about segment version management @himanshug pointed out):

1. When timeline of derived-datasource is less than base-datasource, supervisor will submit a derived-datasource-index-task. The derived-datasource-index-task is a hadoop-index-task. The only difference is that the version of segments generated by derived-datasource-index-task is not the time when task is submitted, but the related base datasource segments version.
2. If the timeline of derived-datasource is more than base-datasource, supervisor will set used=false and submit kill tasks for those segments.
3. Once supervisor find the segment version of derived-datasource is different from its related base-datasource in the same interval, supervisor will set used=false and submit a derived-datasource-index-task for that interval (This idea comes from @jihoonson 's suggestions).

DELETE

When the supervisor is shutdown or resubmited, the previous data of the derived-datasource will be deleted.
When the base datasource is disable, all its derived datasource supervisor will be shutdown.

Optimize query with DerivedDatasource

A MaterialisedViewQueryRunner is added in method applyPreMergeDecoration() of FluentQueryRunnerBuilder, such as

    public FluentQueryRunner applyPreMergeDecoration()
    {
      return from(
          new UnionQueryRunner<T>(
              new MaterialisedViewQueryRunner(
                  toolChest.preMergeQueryDecoration(
                      baseRunner
                  )
              )
          )
      );
    }

In MaterialisedViewQueryRunner, query is rewritten into union queries of derived-datasource and base datasource and merged results of all queries before returnning. The process of optimization is as follows.

1. Check if the datasource is a table datasource. If not, do not optimize.
2. Check if the datasource has derived-datasources, and find the derived-datasources which involve the dimensions and metrics the query need. If no derived datasource meet the condition, do not optimize.
3. Split query interval by segment granularity into sub-intervals, and for each sub-interval, find out the derived-datasource which has the minimum amount of data. Then, replace the query datasource by the derived-datasource in the sub-interval.

In this way, query can be partially covered by the derived datasource and partially by the original one, and the problem @nishantmonu51 pointed out is solved.

[zp-shadowolf]

😁 Thank you very much for your email! The problem has been solved, I restarted IDEA. 2018-01-30 16:51 GMT+08:00 zhangxinyu1 <notifications@github.com>:

…

The motivation of 'materialised view' is in issue #5218 <#5218> . The implementation of 'Materialised View' is diveded into two parts. One is *the management of DerivedDatasource*, and the other is *optimizing query with DerivedDatasource*. Management of DerivedDatasource Each derived-datasource is managed by a derived-datasource supervisor(like kafkaSupervisor). CREATE DerivedDatasource is created when user submits a derived-datasource supervisor. The json file of derived-datasource supervisor should include base datasource name, dimensions and metrics at least. DerivedDatasource name can be setted by users or generated by base datasource name, dimensions and metrics. Segment granularity and query granularity should keep the same with base datasource. The information is stored as metadata of supervisors. MAINTAIN A derived-datasource supervisor should make sure timeline and segment versions are the same with its base datasource (It can sovle the problem about segment version management @himanshug <https://github.com/himanshug> pointed out): 1. When timeline of derived-datasource is less than base-datasource, supervisor will submit a derived-datasource-index-task. The derived-datasource-index-task is a hadoop-index-task. The only difference is that the version of segments generated by derived-datasource-index-task is not the time when task is submitted, but the related base datasource segments version. 2. If the timeline of derived-datasource is more than base-datasource, supervisor will set used=false and submit kill tasks for those segments. 3. Once supervisor find the segment version of derived-datasource is different from its related base-datasource in the same interval, supervisor will set used=false and submit a derived-datasource-index-task for that interval (This idea comes from @jihoonson <https://github.com/jihoonson> 's suggestions). DELETE When the supervisor is shutdown or resubmited, the previous data of the derived-datasource will be deleted. When the base datasource is disable, all its derived datasource supervisor will be shutdown. Optimize query with DerivedDatasource A MaterialisedViewQueryRunner is added in method applyPreMergeDecoration() of FluentQueryRunnerBuilder, such as public FluentQueryRunner applyPreMergeDecoration() { return from( new UnionQueryRunner<T>( new MaterialisedViewQueryRunner( toolChest.preMergeQueryDecoration( baseRunner ) ) ) ); } In MaterialisedViewQueryRunner, query is rewritten into union queries of derived-datasource and base datasource and merged results of all queries before returnning. The process of optimization is as follows. 1. Check if the datasource is a table datasource. If not, do not optimize. 2. Check if the datasource has derived-datasources, and find the derived-datasources which involve the dimensions and metrics the query need. If no derived datasource meet the condition, do not optimize. 3. Split query interval by segment granularity into sub-intervals, and for each sub-interval, find out the derived-datasource which has the minimum amount of data. Then, replace the query datasource by the derived-datasource in the sub-interval. In this way, query can be partially covered by the derived datasource and partially by the original one, and the problem @nishantmonu51 <https://github.com/nishantmonu51> pointed out is solved. — You are receiving this because you are subscribed to this thread. Reply to this email directly, view it on GitHub <#5304>, or mute the thread <https://github.com/notifications/unsubscribe-auth/AeIiBsNt_k-FEtNf8ZYlzttyUPK0HK4Rks5tPtgogaJpZM4Rx9dy> .

[sascha-coenen]

Hi zhangxinyu1,

thank you so much for pushing for this feature. I've been wanting this for a long time and tried to raise awareness for it. This feature is definitely needed. I don't even understand how one can do without it.
I read the comments from the previous proposal.
Let me add a few things to those.

• It was being said that the idea of such a mechanism isn't new and it was mentioned that Vertica has a similar feature named "projections". In Vertica, projections have the same number of rows as the original table. So same same but different. The actual prior art would be OLAP. Many databased call themselves OLAP although they are missing every feature of it. These derived datasources in OLAP terms are called cuboids. And yes, it DOES speed up queries and no it is NOT an option to just add more hardware to the query layer. We have a Druid cluster which is dramatically slow because we need to have many dimensions. If a user makes a wide query and trillions of rows need to be scanned, even a large Druid cluster will need minutes to complete that query. The current caching mechanisms will not help with that. How would I scale up the cluster? If a given query takes one minute to execute I would need to make my cluster 60 times bigger to get the query to subsecond latency.
  It was being said that people are manually reindexing secondary datasources. However, this becomes quite unfeasible very quickly. How to make sure that query results are consistent?
  This heavy lifting should be done by a database system for exactly that purpose that it then isn't required anymore for everyone using the database to do the heavy lifting.

@zhangxinyu1 as I already said, I have invested a lot of thought on bringing this feature to Druid already. I so far always referred to this feature as "Datasource Delegation". Lets stick our heads together to figure this out.
The query path isn't that difficult. One can indeed mix and match which of the many secondary databases to include within a query. That's the beauty of it. An incoming query could be rewritten such that 95% of the timeline of a given query could be covered by the best secondary database, then maybe another 4% is served by segments coming from a bigger secondary database and most likely there will alsways be some segments that have not been re-indexed to some secondary database yet and which therefore would still come from the master datasource.

The most difficult part is to keep track on a per segment granularty basis whether a secondary datasource is in syn with the master. Those timeranges that are not in sync are not elligible to querying.

For the query path, it would also be great if historical queries could contain segment lists coming from different datasources. A historical receive queries that the broker has already rewritten such that there is a list of segments contained in a historical's query. Looking at this segment list, each segment is referenced by its complete ID with the exception of the datasource that the segment belongs to. If one could improve the historical queries such that a list of segments could be passed where every segment also contains the datasource it belongs to, then historicals could already merge data across datasources.
This would totally work: if the broker has determined that a secondary datasource B is up-to-date with master datasouce A for a given segment-granularity slot and therefore picks some segments form B while choosing segments from A for some other time-interval, then a historical could execute a query on both kinds of segments because everything is column-oriented.

Some months back I started implementing this feature but never completed it.

There are other OLAP features that one could nicely layer on top of your proposed solution at a later time. For instance "cell materialization". I once posted a proposal for a different feature which then turned into a discussion of your feature. Perhaps you'd like to read up on it here: https://groups.google.com/forum/#!searchin/druid-development/delegation%7Csort:date/druid-development/Ij3gJjuD4nY/ryC_aCw7GQAJ

[jihoonson]

@zhangxinyu1 first of all, I really apologize for late response. I have been quite busy and forgot about this issue. Sorry.

The updated proposal looks very good to me. Especially the query optimization part sounds good enough as a first step. Regarding the management of derivedDatasources, I have some comments.

Segment granularity and query granularity should keep the same with base datasource.

The segment granularity of the base dataSource can be changed. For example, compactionTask merges all segments in the given interval regardless of whatever their segment granularity was. Keeping query granularity sounds good.

The only difference is that the version of segments generated by derived-datasource-index-task is not the time when task is submitted, but the related base datasource segments version.

I think this might be hacky because it needs to add an exception to our segment versioning system for only derived dataSources. I would think it might be better to keep the version of the base segments which are used for generating the latest derived dataSource. If the version of the base segments is larger than the one in derivedDataSourceSupervisor, the derived dataSource should be disabled and a new task to make the fresh derived dataSource can be submitted.

If the timeline of derived-datasource is more than base-datasource, supervisor will set used=false and submit kill tasks for those segments.

Probably we need to keep a mapping information between original segments and derived segments.

When the base datasource is disable, all its derived datasource supervisor will be shutdown.

I wonder how this can be achieved. Do you have an idea about where the relationships between base dataSources and their derived dataSources are stored?

I think this is a really nice proposal and hope we can make it.

[zhangxinyu1]

@sascha-coenen thanks for your approval.
I'm working on implementing this feature. I'm thinking about how to call it, materialized view selection and maintenance or cuboid selection and maintenance, do you have any suggestions?

For the query path, it would also be great if historical queries could contain segment lists coming from different datasources. A historical receive queries that the broker has already rewritten such that there is a list of segments contained in a historical's query. Looking at this segment list, each segment is referenced by its complete ID with the exception of the datasource that the segment belongs to. If one could improve the historical queries such that a list of segments could be passed where every segment also contains the datasource it belongs to, then historicals could already merge data across datasources.

This is a good idea, It can reduce the cost of data merge on broker. I will try to add this feature in the future.
I will come up with a pr about this feature in few days. Can you help me check it out?
thx.

[zhangxinyu1]

@jihoonson thx for your comments, I will submit a pr in these days. I already implement this function as following:

• About the segment granularity, I generate data of derived datasource based on hadoop-index-task, and the segment granularity is the same with base datasource. There are two reasons why I implement it like this:

1. To make sure the segment versions of base datasource and derived datasource are the same. I used the max(created_date) of segment of base datasource as the segment version of derived datasource. Thus once there are new segment of base datasource created, the max(created_date) will change and derived datasource supervisor will rebuild segments of derived datasource.
2. I think it's easier for derived datasource supervisor to keep the timeline's consistency of base datasource and derived datasource.
   e.g. If base datasource A's segment granularity is hour and derived datasource B's is day, then the change of data during any hour of one day will lead to rebuilding of segment of derived datasource. On the other hand, if A only keeps recently 12 hours' data, it's hard for supervisor to maintain the derived datasource. So I think it's better to make them the same.

• If base datasource is disabled, derived datasource supervisor will not be shut down. But it doesn't matter since the duty of derived datasource supervisor is to keep the consistency of timeline between derived datasource and base datasource. So when base datasource is disabled, derived datasource won't have any data.

[jihoonson]

@zhangxinyu1 it sounds great that you're already working on this! Your explanation sounds good to me overall, but I have one question.

To make sure the segment versions of base datasource and derived datasource are the same. I used the max(created_date) of segment of base datasource as the segment version of derived datasource.

Does this mean that the segment version of base dataSource and derived dataSource are literally same? Would you elaborate more on how you could do this? I'm asking because our segment versioning system is already quite complicated and I hope we don't need to add a sort of special handling for derived dataSource.

[zhangxinyu1]

@jihoonson
Since I can get the max created_date of base datasource segments of specified interval, and it will change as the base segments change, then I use it as the version of derived segment of the specified interval. And I add a derived datasource supervisor to control these. Once the base segment changes, supervisor will find the version of derived segment is different with the max created_date of base one, then the related derived segments will be unused and supervisor will submit a hadoop-index-task with the information of all needed related segments of base datasource to generate the new derived segments.

[jihoonson]

@zhangxinyu1 ah sorry! My question was a little bit vague.

Since I can get the max created_date of base datasource segments of specified interval, and it will change as the base segments change, then I use it as the version of derived segment of the specified interval.

My question is, how will you implement the segmentAllocator to use the same segment version of the base dataSource for the corresponding derived dataSource. Do you assume that any types of tasks can't overwrite/append the derived dataSources unless they are submitted by the derived dataSource supervisor? I think it might be a valid assumption, and if so, segmentAllocator can be implemented to allocate segments based on only supervisor's local information without using IndexerSQLMetadataStorageCoordinator.

[zhangxinyu1]

@jihoonson
Yes, any types of tasks can overwrite/append the derived dataSource. Derived datasource supervisor submit hadoop-index-tasks to generate data. In the HadoopTuningConfig of hadoop-index-task, version is the max created_date and 'useExplicitVersion' is true.

[jihoonson]

@zhangxinyu1 oh I see. That would work.

But, if possible, I think it would be better to use our existing versioning system (e.g., ActionBasedSegmentAllocator) rather than separately maintaining segment versions in derived dataSource supervisors. This is because segment version management can be done in one class (TaskLockBox) rather than being spread over different classes depending on ingestion types. Another reason is that only Hadoop index task supports explicitly specifying segment versions in the task spec. As a result, users should have a Hadoop cluster to use this cool feature.

Do you think it's possible? I guess maybe it is if we keep the segment versions of the original dataSource corresponding to the current derived dataSources in the supervisor. Once the actual segment version of the original dataSource is higher than the one kept in the supervisor, then it is triggered to rebuild derived dataSources.

[zhangxinyu1]

@jihoonson Thanks for your suggestions

Derived dataSource supervisor would not store any segment versions. It's stateless. The things supervisor do is:

1. Reading the max created_date of base dataSource of one interval from db.
2. Reading the version of derived dataSource of the interval from db.
3. compare the max created_date and the version. if they are the same, do nothing. otherwise, submit a task with bases datasource DataSegment to rebuild derived dataSource for the interval.
   In this way, derived dataSource supervisor needn't store any information about the segment version.

This is because segment version management can be done in one class (TaskLockBox) rather than being spread over different classes depending on ingestion types.

Could you please explain why using TaskLockBox to management segment version is better than use existing tasks.

But, you'r right, only hadoop-index-task supported is not enough. I think batch ingestion is suitable for derived datasource rebuilding, and the current batch ingestion can run on peon and run on hadoop cluster. Maybe we should implement one materialized-view-task to support both kinds of ingestion and use the existing version system in the next version.

if we keep the segment versions of the original dataSource corresponding to the current derived dataSources in the supervisor. Once the actual segment version of the original dataSource is higher than the one kept in the supervisor, then it is triggered to rebuild derived dataSources.

If the base datasource segments are generated by kafka-index-service, using the original dataSource version as the derived dataSource version may have some problems. This is because kafka-index-task can generate new segments with the same version. These new segments with the same version would not trigger derived datasource rebuilding.

[jihoonson]

@zhangxinyu1 thanks for the detailed explanation.

Derived dataSource supervisor would not store any segment versions. It's stateless.

Yes, I see your point. But actually the algorithm is stateful because it works based the states (created_date and version) stored in metastore. However, this doesn't always work. Druid's segment versioning system has a rule that recent segments have higher versions than old segments. Using time as the segment version usually works, but we still consider using something else if there is an issue in the future. created_time doesn't have such a rule. As a result, recent segments might have older created_time than old segments depending on overlord's wall clock time.

If you want to make the derived dataSource supervisor stateless, I think it should be something different from the proposed one. One possible solution is triggering updates on derived dataSources whenever a base dataSource segment is published. I already commented here about this idea even though it was too rough. So, I'll explain more details. Publishing segments is completed only after an overlord updates the metadata of those segments in metastore. Tasks should use SegmentTransactionalInsertAction to do it. I think we can add a something new feature like triggerAction which is triggered upon a successful segment insertion in SegmentTransactionalInsertAction. The triggerAction should be a general abstraction of any action. To update derived dataSources, tasks can specify a triggerAction which notifies the published segment information to derived dataSource supervisors. Once a supervisor receives an event from the triggerAction, it can submit a task to update its derived dataSource. Since the segment management is done by coordinators, the triggerAction should be able to be called in coordinators too. In this way, the supervisors don't have to check any state like comparing the created_date of base dataSource segments and the version of derived dataSource segments.

But, I think it's quite complicated and it's always a good idea to start with a simple implementation. If you want to implement a good stateless system for derived dataSource, it would be great. However, if it's not easy, I would say it's better to start with a simpler one which can be easily improved in the future. IMO, using created_date as the segment version might be simple to implement, but is not easy to improve in the future because it adds an assumption that the ordering of created_date of segments is guaranteed. I don't think this assumption should be guaranteed.

Could you please explain why using TaskLockBox to management segment version is better than use existing tasks.

All types of tasks use LockTryAcquireAction or LockAcquireAction which internally get a lock using TaskLockBox. And the acquired lock has a valid segment version. All task types are supposed to use this version.

Specifying a user-customized version in tasks (like Hadoop task) is dangerous. The result segments might not be visible or, even worse, overwrite existing segments unexpectedly. As a result, this should be used very carefully.

Maybe this is not an issue for derived dataSources if they can be created/updated/deleted via only their supervisors (BTW, is this the assumption you have?). However, I'm still not sure about using the created_time as the segment version.

Maybe we should implement one materialized-view-task to support both kinds of ingestion and use the existing version system in the next version.

That makes sense if needed. Would you elaborate a little bit more on the new task type you think?

If the base datasource segments are generated by kafka-index-service, using the original dataSource version as the derived dataSource version may have some problems. This is because kafka-index-task can generate new segments with the same version. These new segments with the same version would not trigger derived datasource rebuilding.

Thanks. I missed Kafka indexing service.

[zhangxinyu1]

@jihoonson thank you so much for explaining the plan of segment version.

Yes, starting with a simple implementation is better, and we can impove it later. Using created_time as segment version is not good, but i can't thing of a better way to implement it. do you have any ideas?

I'm still wondering why the specified segment version is not allowed in TaskLockbox. I think it is flexible to use the version generated by TaskLockBox automatically by default and also allow users to set specified versions when they need. It can be considered as one of advanced usages. if users choose to use specified versions, they should know the results.

Maybe this is not an issue for derived dataSources if they can be created/updated/deleted via only their supervisors (BTW, is this the assumption you have?).

In current implementation, derived dataSource is normal datasource, it can be changed by setting segment rules, submit a task by users, or other operations. However, supervisor always try to keep the timeline's consistency of base datasource and derived datasource.

About the materialized-view-task, I think it should be able to run on peon or hadoop at least. And it depends on the users' config.