[Proposal] 'Materialised View' with pruned dimensions

[zhangxinyu1]

Motivation

In our experience, the number of dimensions of a datasource typically range from a dozen to dozens. However, queries about the datasource always only involve several dimensions. As we know, the fewer dimensions a datasource has, the less time a query cost. Based on the idea of changing time with space, druid can build a derived datasource which only contains several common dimensions of the datasource. When users query the datasource, druid calculates results based on the derived datasource if the derived datasource contains all dimensions the query required.

From another perspective, it is also a compromise between druid and kylin. Druid has minimal degree of pre-calculation, while kylin has maximum degree of pre-calculation. That is, druid only pre-calculation the combined value of all dimensions as shown in Table1(assume there are two dimensions and one metrics). However, kylin pre-calculation all possible combinations of dimensions as shown in Table 1-3. As a result, druid calculates for each query, while kylin stores much useless result data.

Table 1:

Dimension A	Dimension B	Metrics A
a1	b1	1
a2	b1	3
a1	b2	5

Table 2:

Dimension A	Metrics A
a1	6
a2	3

Table 3:

DimensionB	Metrics A
b1	4
b2	5

Virtual datasource is proposed for druid to do more pre-calculation. The datasource which user ingests is called base-datasource, and the datasource which generated based on the base-datasource is called derived-datasource which is very similar to the notion of materialized views in traditional relational databases. Derived-datasource only involves some of dimensions of the base-datasource. It is noteworthy that user only need to know the base-datasource name. When user query the base-datasource, druid can automatically change the datasource of the query to a derived-datasource if the derived-datasource match some conditions, such as including all dimensions the query required.

In this version, we focus on the datasource which is loaded from files.

Implementation

Create and Delete Derived-datasource

Conditions:

1. The timeline, metrics and granularities of a derived-datasource and its base-datasources are the same
2. Derived-datasource dimensions is a subset of base-datasource dimensions.

Implementation

1. Add two http interface in DatasourceResource.java. One is a POST request used to create derived datasource, and the other is a GET request to get information of derived-datasource.
   curl -X POST -d @dimensions.json http://localhost:8081/druid/coordinator/v1/datasources/wikiticker/derivatives -H 'Content-Type:application/json'
   Dimensions.json stored all dimensions of derived-datasource, such as
   ["metroCode","namespace","page","regionIsoCode","regionName","user"]
curl -X GET http://localhost:8081/druid/coordinator/v1/datasources/wikiticker/derivatives -H 'Content-Type:application/json'
   The return result is all derived-datasources of wikiticker, and related dimensions:

{
    "baseDataSource": "wikiticker",
    "derivedDataSource": {
        "wikiticker-0c224343": [
            "metroCode",
            "namespace",
            "page",
            "regionIsoCode",
            "regionName",
            "user"
        ]
    }
}

2. Add an new class DerivedDatasourceManager, which is responsible for:

   • Read and Write information of derived-datasource from database;
   • Traverse all derived-datasource. If the timeline of derived-datasource is less than the base datasource, a hadoop-reindex-task will be submit to ingest missing data. On the contrary, if the timeline of derived-datasource is more than the base datasource, druid will set used=false for the excess segments and a kill task will be submit to remove the data.

3. Add a new table "druid_derivatives" to store information of derived-datasource. The table include 3 columns: basedatasource, deriveddatasource,dimensions. The primary key is combination of basedatasource and dimensions

4. For the purpose of submitting hadoop-reindex-tasks when timeline of derived-datasource is less than base-datasource, it is necessary to get metrics and granularities of base-datasource. Therefore, a table "druid_dataschema" is required to add. It includes 5 columns: basedatasource, start, end, ts, datashcema. The primary key is the combination of basedatasource, start, end and ts.

5. In HadoopIndexTask.java, after publishing segments to database, the schema of the task should be inserted into table "druid_dataschema". Besides, if the datasource of the task has derived datasource and the data of the derived-datasource in the interval of the task has already existed, it’s required to delete this data of derived-datasource.

6. When user delete a datasource, derived-datasource of the datasource should be deleted.

Optimizing Datasource

Conditions:

Once druid receives a query, the datasource of the query is replaced by a derived-datasource(Figure 1). The alternatives must meet the following conditions:

1. The intersection between base-datasource timeline and query interval is equal to the intersection between derived-datasource timeline and query interval.
2. derived-datasource includes all dimensions query need
3. The chosen derived-datasource has the minimum amount of data among all derived-datasources which meet condition 1 and 2.
4. If there is no suitable derived-datasource, base-datasource will not be replaced.
   Figure 1:
   

Implementation

In BrokerResource.java, when receive a query, first optimize datasource: replacing datasource with a suitable derived-datasource. The process of optimization is as follows.

1. Check if the datasource is a table datasource. If not, do not optimize. (In this version, we only support table datasource optimization)
2. Check if the datasource has derived-datasource. If it has, get the information of these derived-datasources: name, dimensions and timeline.
3. Traverse all derived-datasource, and find the derived-datasource which meet condition 1 and 2. If there are some derived-datasources meeting these two conditions, do the 4th step.
4. Find the derived-datasource which has the minimum amount of data among all derived-datasources which meet condition 1 and 2.

[zhangxinyu1]

@gianm @leventov
What do you think about this proposal?

[zhangxinyu1]

@jon-wei @himanshug Could you please give me some suggestions?

[leventov]

Pruning dimensions doesn't win you anything because when querying historical data, unused dimensions are neither accessed in memory nor loaded from disk.

This is the primary advantage of columnar storage. If it wasn't the case, there would be no point in Druid neither any other columnar data store to choose this architecture.

[leventov]

As a result, druid calculates for each query

Side note: enable caching (druid supports both historical- and broker-level query results cache) to reduce the amount of repetitive calculations.

[zhangxinyu1]

@leventov Thanks for the new title.

Pruning dimensions doesn't win you anything because when querying historical data, unused dimensions are neither accessed in memory nor loaded from disk.

Pruning dimensions can reduce the number of rows. For example, there are two dimensions A and B in a segment. The cardinality of A is 10, and the cardinality of B is 100. The max number of rows in this segment is 10*100. However, if there is only dimension A in this segment, the max number of rows in is 10.

[sixtus]

You can already do it via union queries. I don’t think it‘s worth adding it to druid itself.

[zhangxinyu1]

@leventov

Side note: enable caching (druid supports both historical- and broker-level query results cache) to reduce the amount of repetitive calculations.

Indeed, query caching can reduce the amount of repetitive calculations. It stores the results on a per segment basis and a given query. This proposal pre-calculates the results based on dimensions and the results are considered as a datasource. We can calculate any query (with given dimensions) based on the datasource instead of a given query.

[zhangxinyu1]

@sixtus
I don't know how to do it via union queries. Could you please explain it？ Thanks.

[leventov]

Pruning dimensions can reduce the number of rows. For example, there are two dimensions A and B in a segment. The cardinality of A is 10, and the cardinality of B is 100. The max number of rows in this segment is 10*100. However, if there is only dimension A in this segment, the max number of rows in is 10.

Could you explain what do you mean by "row" here? "Number of rows" in Druid terms could be "maximum" or "minimum", it's just static, and independent of dimensions's cardinalities.

[nishantmonu51]

I think what @zhangxinyu1 wants to point out here is that you will get better rollup when you have fewer dimensions with less cardinality, which means less rows to scan at query time which is expected to lead to better query performance.

Havn't really thought through all the proposed implementation details, Few initial questions -

1. How would segment versioning work between derived datasource and normal datasource ?
2. What would be behavior for queries for which the interval is partially covered by the derived datasource and partially by the original one ? Will they be served fully by original or derived datasource? I expect it would be a very common case for realtime ingestion.
3. are you considering doing that via an extension or a core feature ? I think an extension would be better option.

[zhangxinyu1]

Let me use a more detail example to explain it.
Given a datasource with dimension A and B, such as:

Dimension A	Dimension B	Metrics A
a1	b1	1
a2	b1	3
a1	b2	5
a2	b2	7

In this datasource, there are 4 rows.

If we prune Dimension B, we can get the datasource:

Dimension A	Metrics A
a1	6
a2	10

There are only 2 rows in this datsource.

Then, we want to calculate select sum(Metrics A) where DimensionA='a1'. If we use the first datasource, druid need read the value of MetricsA in the first row and the 3rd row, and add them. But if we use the second datasource, druid only need read the value of MetricsA in the first row.

[leventov]

Ok, I understand now. It specializes for a specific aggregation, sum in your example.

I think your proposition with "transparent" data source names and special handling on historical query side is too complex. IMO it would be more reasonable to add handling of this on ingestion / batch indexing side, by adding some type of "multi data source ingestion spec", with multiple parts, where each part says the name of the output data source (not related to others, when segments are produced), dimensions to prune, and type of rollup aggregation (if any) which should be applied.

[zhangxinyu1]

@nishantmonu51

I think what @zhangxinyu1 wants to point out here is that you will get better rollup when you have fewer dimensions with less cardinality, which means less rows to scan at query time which is expected to lead to better query performance.

Yes, that's exactly what I mean. Thanks.

1. How would segment versioning work between derived datasource and normal datasource ?

In the same interval, segment versions of derived datasrouces are always after normal datasources. In the same interval, if the segment version of derived datasource is before the related normal datasource, the derived datasource in that interval should be reindexed.

2. What would be behavior for queries for which the interval is partially covered by the derived datasource and partially by the original one ? Will they be served fully by original or derived datasource? I expect it would be a very common case for realtime ingestion.

It's a good question! I hope the query can be query can be served partially by derived datasource and partially by original. However, in this implementation, it's only served fully by original. Therefore, in this implementation, it cannot support realtime ingestion.

3. are you considering doing that via an extension or a core feature ? I think an extension would be better option.

Yes, I am considering do it via an extension. As far as I can see, some interfaces need to be added into core, is it ok?

[jihoonson]

@zhangxinyu1 thanks for the proposal. I wonder what are the primary benefits of the derivedDataSource. Is the query speed the only one?

It's a good question! I hope the query can be query can be served partially by derived datasource and partially by original. However, in this implementation, it's only served fully by original. Therefore, in this implementation, it cannot support realtime ingestion.

This can also happen even in batch append ingestion. I think this is important and should be supported because most Druid users will append new data to existing dataSources in some ways.