Support Continuous Query Feature

[yuxianglong]

In Devops/IOT scenes when using time-series databases, qps(queries per second) for database is getting slower when data nodes grow (including numbers of data, the dimensions of the data), also, the longer the data is from now, the less accurate we require them.

Continuous Query(CQ) can provide the feature to aggregate data according to the SQL which defined by user, compare with other down-sample methods(e.g sliding-down-sample), CQ can support more SQL which are more complex and flexible, take continuous query in InfluxDB as an example:

data

name: bus_data
--------------
time                   passengers   complaints
2016-08-28T07:00:00Z   5            9
2016-08-28T07:15:00Z   8            9
2016-08-28T07:30:00Z   8            9
2016-08-28T07:45:00Z   7            9
2016-08-28T08:00:00Z   8            9
2016-08-28T08:15:00Z   15           7
2016-08-28T08:30:00Z   15           7
2016-08-28T08:45:00Z   17           7
2016-08-28T09:00:00Z   20           7

create continuous query:

CREATE CONTINUOUS QUERY "cq_basic" ON "transportation"
BEGIN
  SELECT mean("passengers") INTO "average_passengers" FROM "bus_data" GROUP BY time(1h)
END

cq_basic_rp executes at one-hour intervals, the same interval as the GROUP BY time() interval. Every hour, cq_basic_rp runs a single query that covers the time range between now() and now() minus the GROUP BY time() interval, that is, the time range between now() and one hour prior to now().

Annotated log output on the morning of August 28, 2016:

At 8:00 cq_basic_rp executes a query with the time range time >= '7:00' AND time < '8:00'.
cq_basic_rp writes one point to the three_weeks RP and the average_passengers measurement:

>
    name: average_passengers
    ------------------------
    time                   mean
    2016-08-28T07:00:00Z   7
>

At 9:00 cq_basic_rp executes a query with the time range
time >= '8:00' AND time < '9:00'.
cq_basic_rp writes one point to the three_weeks RP and the average_passengers measurement:

>
    name: average_passengers
    ------------------------
    time                   mean
    2016-08-28T08:00:00Z   13.75

Results:

> SELECT * FROM "transportation"."three_weeks"."average_passengers"

name: average_passengers
------------------------
time                   mean
2016-08-28T07:00:00Z   7
2016-08-28T08:00:00Z   13.75

[shilinlee]

CQ development plan design

---

[TOC]

precondition

In the case of node restart or network failure, accept a certain degree of repeated execution of CQ.

Parse

#257

Create

Create continues query through ts-sql

CREATE CONTINUOUS QUERY <cq_name> ON <database_name>
RESAMPLE EVERY <interval> FOR <interval>
BEGIN
  <cq_query>
END

• CQ limits: concurrent exec continues queries goroutines number.

[continuous_queries]
# concurrent exec continues queries goroutines number
max-process-CQ-number = 5

• <cq_name> should be globally unique.
• Check whether the target <db>,<rp> exists.
• Check whether the target measurement exists and whether the measurementconflicts with the current CQ.
• The output of a continuous query must go to either a different measurement or to a different retention policy.
• In theory they'd still be able to create a cycle with multiple continuous queries. We should check for these and disallow.
• The CQ execution frequency cannot be less than 1s.

ts-meta persistence continues query task

openGemini/open_src/influx/meta/database.go

Lines 18 to 24 in 8674137

	type DatabaseInfo struct {
	Name string
	DefaultRetentionPolicy string
	RetentionPolicies map[string]*RetentionPolicyInfo
	MarkDeleted bool
	ShardKey ShardKeyInfo
	}

type DatabaseInfo struct {
	......
  ContinuousQueries []ContinuousQueriesInfo # ⭐️
}

type ContinuousQueriesInfo struct {
  Name string
  Query string
}

• The ContinuousQueries field is added, please ensure upgrade compatibility.

additional

• every time CQ completes a cycle, it needs to check whether the task is deleted. You also need to check whether the load can be balanced to other SQL nodes.

Run

unique ID: IP:PORT

SQL -> Meta heartbeat every second.

Meta assigns different CQ tasks to different SQL.

• Each SQL node reports LastRunTime to the Meta after all CQ tasks on the node are complete.

Scheduling

1. Meta restart:
   • Through the heartbeat of sql->meta, we know how many SQLs there are and their IDs.
2. SQL Stopped:
   • If Meta cannot detect a certain CQ for more than X cycles, the CQ task will be recycled.
   • Meta assigns these tasks to other SQL
   • The ID of this SQL is also deleted from Meta, or marked as a tombstone.
3. SQL Restart:
   • SQL ID：If the SQL ID is stored in the Meta memory, it will be returned directly.
   • CQ Tasks: Communicate with meta and let meta assign tasks.

Query task

> SHOW CONTINUOUS QUERIES

name: _internal
name    query
----    -----
cq_demo  CREATE CONTINUOUS QUERY ......

Get CQ tasks from Meta.

Stop

DROP CONTINUOUS QUERY <cq_name> ON <database_name>

1. Delete the CQ task from Meta. Save CQ records to tombstone.
2. SQL detects that CQ has been deleted, so it will not be executed in the next cycle.

Trigger

Support manually triggering a CQ to execute once through the curl command.

Monitor

Consider adding monitoring items, such as quantity, execution time, whether to report errors, etc.

[scuzyj]

1、The number of CQs must be limited. Number of DataNode x 100. --- what type of dataNode is, SQL or Store Node？
2、What is policy of cq task scheduling? The least task is preferred?
What are the handling policies in the following scenarios? After the SQL process is restarted or a new SQL process is added.

[shilinlee]

1. This strategy is changed to limit the number of concurrent processing of CQ tasks.

[continuous_queries]
# concurrent exec continues queries goroutines number
max-process-CQ-number = 5

2. SQL restart or new addition requires meta to re-evaluate whether the CQ tasks are balanced