risingwavelabs / risingwave Goto Github PK

Currently, we simply print the error as warning in console, and let the actor down. In the future, we should report this to dashboard, and restart the actor.

At the same time, some of our implementations also spawns extra futures apart from the actor. We should also manage this. For example, BarrierAligner will spawn two futures to poll data from each of the executors. #2229

We introduce the concept of transaction in Hummock to aim the checkpointing process so that changes can be committed and rollbacked atomically.

• Implement Hummock service atop meta service and migrate version manager (#1744 singularity-data/risingwave-legacy#2156)
• Introduce transaction in hummock service (transaction state, lifecycle, visibility management) (#2513)
• Support transaction commit (#2513)
• Support transaction rollback (#2513)
• State store integration
• Stream manager integration

Blockers:

• singularity-data/risingwave-legacy#2263

Query:

select
	sum(l_extendedprice* (1 - l_discount)) as revenue
from
	lineitem,
	part
where
	(
		p_partkey = l_partkey
		and p_brand = ':1'
		and p_container in ('SM CASE', 'SM BOX', 'SM PACK', 'SM PKG')
		and l_quantity >= :4 and l_quantity <= :4 + 10
		and p_size between 1 and 5
		and l_shipmode in ('AIR', 'AIR REG')
		and l_shipinstruct = 'DELIVER IN PERSON'
	)
	or
	(
		p_partkey = l_partkey
		and p_brand = ':2'
		and p_container in ('MED BAG', 'MED BOX', 'MED PKG', 'MED PACK')
		and l_quantity >= :5 and l_quantity <= :5 + 10
		and p_size between 1 and 10
		and l_shipmode in ('AIR', 'AIR REG')
		and l_shipinstruct = 'DELIVER IN PERSON'
	)
	or
	(
		p_partkey = l_partkey
		and p_brand = ':3'
		and p_container in ('LG CASE', 'LG BOX', 'LG PACK', 'LG PKG')
		and l_quantity >= :6 and l_quantity <= :6 + 10
		and p_size between 1 and 15
		and l_shipmode in ('AIR', 'AIR REG')
		and l_shipinstruct = 'DELIVER IN PERSON'
	);

Currently, we only test our debug build of our program. In the future, we should use cargo build --release for our e2e test. This is the convention followed by TiKV: run unit tests with debug version, and integration tests with release version.

Currently we've implemented almost all necessary operators and expressions required by TPC-H. This issue tracks bug-fix and implementation of TPC-H queries.

A query is marked as supported iff. it has been added to the end-to-end test.

SQL Features in Queries

Batch

• Batch: TPC-H Q1
• #177
• #439
• #296
• #440
• Batch: TPC-H Q6
• #441
• #443
• singularity-data/risingwave-legacy#2392
• singularity-data/risingwave-legacy#2393
• singularity-data/risingwave-legacy#2394
• #217
• singularity-data/risingwave-legacy#2422
• singularity-data/risingwave-legacy#2423
• #272
• singularity-data/risingwave-legacy#2426
• singularity-data/risingwave-legacy#2427
• singularity-data/risingwave-legacy#2428
• #276
• #277
• #278
• #279

Streaming

• Streaming: TPC-H Q1
• #178
• singularity-data/risingwave-legacy#1907
• #179
• singularity-data/risingwave-legacy#1908
• Streaming: TPC-H Q6
• singularity-data/risingwave-legacy#2638
• singularity-data/risingwave-legacy#2644
• Streaming: TPC-H Q9
• singularity-data/risingwave-legacy#2556
• #490
• singularity-data/risingwave-legacy#2839
• singularity-data/risingwave-legacy#2840
• singularity-data/risingwave-legacy#2890
• #496
• #491
• singularity-data/risingwave-legacy#2897
• singularity-data/risingwave-legacy#1911
• #497
• #492
• #495
• #494

Basically there are two ways to support it:

    - use: compute-node
      port: 5689
      exporter-port: 1224
      state-backend: s3://bucket

If using the above way, we will need to add state-backend parameter to every compute-node. In some cases, we need to write it 3 times.

Another way is to make s3 as a mock service.

  default:
    - use: s3
       bucket: xxxx
    - use: meta-node
    - use: compute-node
    - use: prometheus
    - use: frontend

... and for compute-node:

  compute-node:
    address: "127.0.0.1"
    port: 5688
    exporter-address: "127.0.0.1"
    exporter-port: 1222
    id: compute-node-${port}
    provide-minio: "minio*"
    provide-meta-node: "meta-node*"
+   provide-s3: "s3*"
    user-managed: false

Query:

select
	cntrycode,
	count(*) as numcust,
	sum(c_acctbal) as totacctbal
from
	(
		select
			substring(c_phone from 1 for 2) as cntrycode,
			c_acctbal
		from
			customer
		where
			substring(c_phone from 1 for 2) in
				(':1', ':2', ':3', ':4', ':5', ':6', ':7')
			and c_acctbal > (
				select
					avg(c_acctbal)
				from
					customer
				where
					c_acctbal > 0.00
					and substring(c_phone from 1 for 2) in
						(':1', ':2', ':3', ':4', ':5', ':6', ':7')
			)
			and not exists (
				select
					*
				from
					orders
				where
					o_custkey = c_custkey
			)
	) as custsale
group by
	cntrycode
order by
	cntrycode;

Reference:

• Design note: https://singularity-data.larksuite.com/docs/docusUkBFoG9B5CwOruT68Sx65f
• End to end benchmark roadmap: #2232

Tracking:
Stage one (target 2022.1.5):

• A command line executable tool ss_bench with following capabilities:
  • Configurable backends: (InMemoryStateStore, hummock+s3, hummock+minio, TikvStateStore, RocksDBStateStore)
  • Configurable hummock options: (SST size, block size, bloom filter, default bucket, checksum algo, etc..)
  • Configurable data sizes: (key length/value length).
  • Configurable mode: (simple get/ simple scan / simple write_batch / mixed).

Stage two (target 2022.1.10):

• Launch Premetheus endpoint to expose internal storage counters.
• Configurable concurrencies: (multithreaded get/scan/write_batch operations).
• Review performance with ss_bench.

Stage three: more likely patterns as our internal states:

• Sequencial writes within single key space.
• Random writes within single key space.
• (Sequencial, random) x (writes, deletes, updates) within single key space.
• (Sequencial, random) x (writes, deletes, updates) x (single key space, multiple key spaces)
• (Tiered, Leveled compaction) x (Sequencial, random ingestion) x (single, multiple key space)

Binder (validator) to bind & check variables exist in their namespace

Clauses:

• Values
• #100
• #174
• #302
• #1278
• #898
• #430
• #633
• #641
• #2166
• #909
• #1167
• #919
• #992

Types & Exprs:

• Bind data types. @neverchanje
• Bind cast, and check if value can be castible to the type.
• #724
• #847
• #1343

Query:

select
	s_suppkey,
	s_name,
	s_address,
	s_phone,
	total_revenue
from
	supplier,
	(
		select
		l_suppkey,
		sum(l_extendedprice * (1 - l_discount))
	from
		lineitem
	where
		l_shipdate >= date '1993-01-01'
		and l_shipdate < date '1993-01-01' + interval '3' month
	group by
		l_suppkey
	) as revenue0 (supplier_no, total_revenue)
where
	s_suppkey = supplier_no
	and total_revenue = (
		select
			max(total_revenue)
		from
			(
				select
				l_suppkey,
				sum(l_extendedprice * (1 - l_discount))
			from
				lineitem
			where
				l_shipdate >= date '1993-01-01'
				and l_shipdate < date '1993-01-01' + interval '3' month
			group by
				l_suppkey
			) as revenue0 (supplier_no, total_revenue)
	)
order by
	s_suppkey;

"Single mode" was designed to generate and execute a plan without Exchange, which is only used in early development.

Note that the so-called "single mode" is NOT a mode designed for single node deployment. Exchange is necessary for parallel execution no mater on single node or cluster.

Currently, there is more and more code handling "single mode". It's time to remove these stuff.

There're a lot in common between a TableSouce and MView. In this tracking issue, we will unify their write paths and integrate batch table with Hummock storage. For more information, check this design doc.

• singularity-data/risingwave-legacy#2412
• singularity-data/risingwave-legacy#2391
• singularity-data/risingwave-legacy#2464
• singularity-data/risingwave-legacy#2492
• singularity-data/risingwave-legacy#2584
• Wait for #87 to be ready to support use TableSourceV2 as streaming source.
• #212
• #311
• #387
• #423
• Bug fixes...

This is to track the works related to benckmarking state store of RisingWave. The final goal is to do the experiments in this doc: https://singularity-data.larksuite.com/docs/docusF6l8lV5BuOCWoq7Cu71fEe

Stage 1

• Dimensions and metrics for state management of RisingWave
• Experiments configuration
• Implement state backends for RocksDB singularity-data/risingwave-legacy#2292
• Implement state backends for TiKV singularity-data/risingwave-legacy#2202
• Implement MetricsManager for RisingWave singularity-data/risingwave-legacy#2198 singularity-data/risingwave-legacy#2234

Stage 2

• singularity-data/risingwave-legacy#2285 singularity-data/risingwave-legacy#2359
• singularity-data/risingwave-legacy#2410
• singularity-data/risingwave-legacy#2436
• #195
• #194
• singularity-data/risingwave-legacy#2411
• #45 (Implement ss_bench tool for state store layer)
• #127

Stage 3

• Conduct StateStore benchmark using ss_bench on memory, RocksDB, TiKV and Hummock backends
• E2E Performance experiments(TPC-H)
• E2E Performance experiments(Nexmark)
• Overhead & cost experiments

We should know the corresponding SQL and plan for a given group of fragment in dashbaord.

Reference: PostgreSQL: Documentation: 14: Chapter 9. Functions and Operators

General for all types

• TEXT format input / output (cast from / into string)
• BINARY format input / output
• singularity-data/risingwave-legacy#1916
• singularity-data/risingwave-legacy#667
• /issues/2284

Aggregate

• count

Equality and Ordering for "most" types

• = / <> / !=
• #2684
• /issues/2283
• singularity-data/risingwave-legacy#666
• < / <= / > / >=
• Implement (NOT) BETWEEN AND functions
• #12761

Aggregate

• min / max

Boolean

• explicit cast from / into int
• Implement AND/OR/NOT operators
• IS [NOT] [TRUE | FALSE]
• #8933

Aggregate

• #8936

Window functions

• #8961

Misc

• #14118

More

• #8806
• #8831
• #8849
• #7117
• #7714

Query:

select
	supp_nation,
	cust_nation,
	l_year,
	sum(volume) as revenue
from
	(
		select
			n1.n_name as supp_nation,
			n2.n_name as cust_nation,
			extract(year from l_shipdate) as l_year,
			l_extendedprice * (1 - l_discount) as volume
		from
			supplier,
			lineitem,
			orders,
			customer,
			nation n1,
			nation n2
		where
			s_suppkey = l_suppkey
			and o_orderkey = l_orderkey
			and c_custkey = o_custkey
			and s_nationkey = n1.n_nationkey
			and c_nationkey = n2.n_nationkey
			and (
				(n1.n_name = 'ROMANIA' and n2.n_name = 'INDIA')
				or (n1.n_name = 'INDIA' and n2.n_name = 'ROMANIA')
			)
			and l_shipdate between date '1995-01-01' and date '1996-12-31'
	) as shipping
group by
	supp_nation,
	cust_nation,
	l_year
order by
	supp_nation,
	cust_nation,
	l_year;

This is a tracking issue for Hummock File Cache System.

Hummock File Cache System servers as a new tier of cache to utilize the spare disk space as block cache.

• Design doc.
• S3 benchmark.
• EBS DIO benchmark
• #3556
• #3889
• ...

Currently, e2e doesn't take Rust codebase into account when calculating coverage. In the future, we should do this.

Benchmark TPC-H Q1: insert statement with 6000+ rows is time consuming

Already tried to suppress logging.
Seems like each inserted value is processed multiple passes during planning/optimization.
(Thanks @zehaowei for the report and investigation.)

Query:

select
	l_shipmode,
	sum(case
		when o_orderpriority = '1-URGENT'
			or o_orderpriority = '2-HIGH'
			then 1
		else 0
	end) as high_line_count,
	sum(case
		when o_orderpriority <> '1-URGENT'
			and o_orderpriority <> '2-HIGH'
			then 1
		else 0
	end) as low_line_count
from
	orders,
	lineitem
where
	o_orderkey = l_orderkey
	and l_shipmode in ('FOB', 'SHIP')
	and l_commitdate < l_receiptdate
	and l_shipdate < l_commitdate
	and l_receiptdate >= date '1994-01-01'
	and l_receiptdate < date '1994-01-01' + interval '1' year
group by
	l_shipmode
order by
	l_shipmode;

Blocking:

• #218
• #257

Query:

select
	o_orderpriority,
	count(*) as order_count
from
	orders
where
	o_orderdate >= date '1997-07-01'
	and o_orderdate < date '1997-07-01' + interval '3' month
	and exists (
		select
			*
		from
			lineitem
		where
			l_orderkey = o_orderkey
			and l_commitdate < l_receiptdate
	)
group by
	o_orderpriority
order by
	o_orderpriority;

Blocker:

• #1916
• #154
• singularity-data/risingwave-legacy#2316

Query:

select
	s_name,
	s_address
from
	supplier,
	nation
where
	s_suppkey in (
		select
			ps_suppkey
		from
			partsupp
		where
			ps_partkey in (
				select
					p_partkey
				from
					part
				where
					p_name like ':1%'
			)
			and ps_availqty > (
				select
					0.5 * sum(l_quantity)
				from
					lineitem
				where
					l_partkey = ps_partkey
					and l_suppkey = ps_suppkey
					and l_shipdate >= date ':2'
					and l_shipdate < date ':2' + interval '1' year
			)
	)
	and s_nationkey = n_nationkey
	and n_name = ':3'
order by
	s_name;

CREATE TABLE num_data (id int4, val numeric(210,10));

numeric test in regress test require numeric type with precision at 210, but the implementation of decimal in https://docs.rs/rust_decimal/latest/rust_decimal/ have only 64 bits

The granularity of cache population and eviction depends on the performance of S3 put / multi part upload / get / part get / byte-range get. We need to bench S3 first before deciding some designs.

#198

Query:

select
	s_name,
	count(*) as numwait
from
	supplier,
	lineitem l1,
	orders,
	nation
where
	s_suppkey = l1.l_suppkey
	and o_orderkey = l1.l_orderkey
	and o_orderstatus = 'F'
	and l1.l_receiptdate > l1.l_commitdate
	and exists (
		select
			*
		from
			lineitem l2
		where
			l2.l_orderkey = l1.l_orderkey
			and l2.l_suppkey <> l1.l_suppkey
	)
	and not exists (
		select
			*
		from
			lineitem l3
		where
			l3.l_orderkey = l1.l_orderkey
			and l3.l_suppkey <> l1.l_suppkey
			and l3.l_receiptdate > l3.l_commitdate
	)
	and s_nationkey = n_nationkey
	and n_name = ':1'
group by
	s_name
order by
	numwait desc,
	s_name;

Blocker:

• #521
• #587

About version in meta rpc. every ddl will make the version of schema +1.

RiseLAB is now capable of all tasks of start_cluster.sh, and the e2e-test-riselab is now running smoothly across main branch and PRs. After we added source support for RiseLAB e2e, it seems to be a good time to retire start_cluster.sh in CI.

• singularity-data/risingwave-legacy#2459
• singularity-data/risingwave-legacy#2460
• #957

Currently when inserting tuples into distributed tables, each partition will allocate implicit ids according to their local row number independently. This is done at singularity-data/risingwave-legacy#1613 .

Duplication may occur in a distributed table when union all partitions together.

To fix this, a quick fix is to assign tuple id as a 16-digit parition id + 48 digit local row id.

Query:

select
	o_year,
	sum(case
		when nation = 'INDIA' then volume
		else 0
	end) / sum(volume) as mkt_share
from
	(
		select
			extract(year from o_orderdate) as o_year,
			l_extendedprice * (1 - l_discount) as volume,
			n2.n_name as nation
		from
			part,
			supplier,
			lineitem,
			orders,
			customer,
			nation n1,
			nation n2,
			region
		where
			p_partkey = l_partkey
			and s_suppkey = l_suppkey
			and l_orderkey = o_orderkey
			and o_custkey = c_custkey
			and c_nationkey = n1.n_nationkey
			and n1.n_regionkey = r_regionkey
			and r_name = 'ASIA'
			and s_nationkey = n2.n_nationkey
			and o_orderdate between date '1995-01-01' and date '1996-12-31'
			and p_type = 'PROMO BRUSHED COPPER'
	) as all_nations
group by
	o_year
order by
	o_year;

MVP Features

• singularity-data/risingwave-legacy#2261
• singularity-data/risingwave-legacy#2572
• singularity-data/risingwave-legacy#2669
• singularity-data/risingwave-legacy#2347
• singularity-data/risingwave-legacy#2473
• #958

Refactor

• singularity-data/risingwave-legacy#2560
• #415
• singularity-data/risingwave-legacy#2592
• singularity-data/risingwave-legacy#2917

Table V2

• #1199

Mainly benchmarking for state management of RisingWave, e2e throughput, price/performance, data freshness, e2e latency

Functionality

• Cmd tool
• Configurable source & workload generator
• Execution pipeline(pg sql, Kafka, CDC)
• Metrics service
• Support using pg SQL & table source to benchmark RisingWave
• Support using Kafka to benchmark RisingWave
• Support using CDC(MySQL->Debezium->Kafka) to benchmark RisingWave

Source

• Implement TPC-H as benchmarking source

Min/Max on boolean types is not supported yet.

CREATE TABLE supplier (
        s_suppkey  INTEGER,
        s_name VARCHAR(25),
        s_address VARCHAR(40),
        s_nationkey INTEGER,
        s_phone VARCHAR(15),
        s_acctbal NUMERIC,
        s_comment VARCHAR(101));

select min(s_suppkey > 1) from supplier;

Currently, Dispatcher only supports single distribution key:

// A dispatcher redistribute messages.
// We encode both the type and other usage information in the proto.
message Dispatcher {
  enum DispatcherType {
    SIMPLE = 0;
    HASH = 1;
    BROADCAST = 2;
  }
  DispatcherType type = 1;
  int32 column_idx = 2; // <--- HERE!!
}

println!("{}", std::mem::size_of::<AllOrNoneState<MemoryStateStore>>());

176

It's too huge as a hash value, but we can optimize it later.

Originally posted by @TennyZhuang in #2114 (comment)

Currently, we can only get output of HashAgg after flush is called. That is because states like Max requires:

1. apply_batch generates a sequence of change log (flush_status)
2. merge the status to the state store by flushing them in write batch
3. read the real max from the state store

This will cause several problems:

• flushing a large incremental state will lead to spike latency
• in some of our fail-over designs, we can only add SST to the LSM tree when all nodes have processed the barrier

Therefore, managed states shouldn't require that all changes have been flushed before producing the correct output for this epoch.

To achieve this, we will need:

• implement iterator on StateStore
• implement state merge on get_output for ManagedExtremeState
• implement state merge on ManagedStringAggState
• ... when other ManagedStates are added, add new tasks here
• support async flush in HashAgg
• support async flush in SimpleAgg
• ... when other executors are integrated into the state store, add new tasks here

RiseLAB currently only kill the tmux session. In the future, we should support graceful exit.

• Call tmux list-windows -a -F "#{pane_pid} #{window_name}" to get pid of the components.
• Send kill -SIGINT -pgid to stop the components.

In pg, the result is :

postgres=# SELECT (-32768)::int2 % (-1)::int2;
 ?column?
----------
        0
(1 row)

In our system, the result is:

dev=>SELECT (-32768)::int2 % (-1)::int2;
Error: Out of range
(1 row)

This also requires the meta service to record information about plans apart from fragments.

This issue tracks the works to rewrite a basic frontend in rust.

This new frontend is expected to be a replacement for the current Java frontend, but they will coexist for several months. During this period, we may need two end-to-end tests for each frond-end respectively, until the new frontend can cover all features.

Expected Features

In the first stage, we only support

• CREATE TABLE
• DROP TABLE
• INSERT
• SELECT
  • columns or constant values (no expressions allowed)
  • from one table (no joins allowed)
  • with very simple WHERE condition (column = constant)
• CREATE MATERIALIZED VIEW
  • Same as SELECT

Components

• Postgres wire protocol: singularity-data/risingwave-legacy#2019
• Server: #293
• Parser: singularity-data/risingwave-legacy#2170
• Binder: #338
• Optimizer: #109
• Catalog: #959
• Execution: #202
• Query Manager: #1015
• #838
• #1219