RisingWave is a cloud-native streaming database that uses SQL as the interface language.

thread 'tokio-runtime-worker' panicked at 'invalid iter', storage/src/hummock/sstable/sstable_iterator.rs:95:14 stack backtrace:

https://github.com/singularity-data/risingwave-dev/runs/5462156952?check_suite_focus=true

What's changed and what's your intention?

This PR is for supporting partial checkpoint proposed in #1157.

In this PR, we add a table_id parameter in ingest_batch, sync and wait_epoch method of trait StateStore so that we are able to keep track of where the kvs are written from. Besides, we group the write batches by their table_id in shared_buffer_uploader and enable syncing a subset of tables in an epoch.

Such concept of table is introduced for finer-grained tracking about where the kvs are written from, so that we are able to support partial checkpoint, vertical grouping and shared arrangement in the future.

A storage table id is usually the operator id, or also the table id of the relational table. Currently the table_id is not assigned yet, and all table_ids are GLOBAL_STORAGE_TABLE_ID for the time being. In future development we will assign table_id according to operator id.

Some unit tests and related structs are modified accordingly.

Checklist

• [x] I have written necessary docs and comments
• [x] I have added necessary unit tests and integration tests

Refer to a related PR or issue link (optional)

#1157

Describe the bug

When running playground on macOS using latest main (first bad commit db6691ba142af74544d208afcd3e47b809b00e00), the following sql commands leads to a server crash with segfault/EXC_BAD_ACCESS.

It works as expected in cluster mode (./risedev d) rather than playground.

To Reproduce

CREATE TABLE t(a int, b int);
CREATE VIEW v AS SELECT * FROM t;
DROP TABLE t;

Expected behavior

Before that commit we were able to see the expected error:

ERROR:  QueryError: Permission denied: PermissionDenied: Fail to delete table `t` because 1 other relation(s) depend on it

Additional context

console warnings before segfault (also there on last good commit, may unrelated):

2022-11-04T15:37:18.08988+08:00  WARN risingwave_storage::hummock::state_store: sealing invalid epoch    
2022-11-04T15:37:18.090268+08:00  WARN risingwave_storage::hummock::state_store: syncing invalid epoch    

backtrace from lldb:

* thread #7, name = 'risingwave-main', stop reason = EXC_BAD_ACCESS (code=1, address=0x0)
  * frame #0: 0x000000019f39685c libunwind.dylib`libunwind::CFI_Parser<libunwind::LocalAddressSpace>::parseFDEInstructions(libunwind::LocalAddressSpace&, libunwind::CFI_Parser<libunwind::LocalAddressSpace>::FDE_Info const&, libunwind::CFI_Parser<libunwind::LocalAddressSpace>::CIE_Info const&, unsigned long, int, libunwind::CFI_Parser<libunwind::LocalAddressSpace>::PrologInfo*) + 204
    frame #1: 0x000000019f396710 libunwind.dylib`libunwind::UnwindCursor<libunwind::LocalAddressSpace, libunwind::Registers_arm64>::getInfoFromFdeCie(libunwind::CFI_Parser<libunwind::LocalAddressSpace>::FDE_Info const&, libunwind::CFI_Parser<libunwind::LocalAddressSpace>::CIE_Info const&, unsigned long, unsigned long) + 100
    frame #2: 0x000000019f3963e8 libunwind.dylib`libunwind::UnwindCursor<libunwind::LocalAddressSpace, libunwind::Registers_arm64>::getInfoFromDwarfSection(unsigned long, libunwind::UnwindInfoSections const&, unsigned int) + 184
    frame #3: 0x000000019f3962a0 libunwind.dylib`libunwind::UnwindCursor<libunwind::LocalAddressSpace, libunwind::Registers_arm64>::setInfoBasedOnIPRegister(bool) + 1012
    frame #4: 0x000000019f398788 libunwind.dylib`libunwind::UnwindCursor<libunwind::LocalAddressSpace, libunwind::Registers_arm64>::step() + 696
    frame #5: 0x000000019f39b138 libunwind.dylib`_Unwind_Backtrace + 352
    frame #6: 0x0000000109081e40 risingwave`std::backtrace::Backtrace::create::h908375f7f84cb508 [inlined] std::backtrace_rs::backtrace::libunwind::trace::h471a59e08ff9e5dc at mod.rs:66:5 [opt]
    frame #7: 0x0000000109081e30 risingwave`std::backtrace::Backtrace::create::h908375f7f84cb508 [inlined] std::backtrace_rs::backtrace::trace_unsynchronized::h4e694232d85e2708 at mod.rs:66:5 [opt]
    frame #8: 0x0000000109081e24 risingwave`std::backtrace::Backtrace::create::h908375f7f84cb508 at backtrace.rs:333:13 [opt]
    frame #9: 0x00000001060d8454 risingwave`_$LT$risingwave_meta..error..MetaError$u20$as$u20$core..convert..From$LT$risingwave_meta..error..MetaErrorInner$GT$$GT$::from::hb4b62fbc8685e728(inner=<unavailable>) at error.rs:66:33
    frame #10: 0x0000000105e7e1d8 risingwave`_$LT$T$u20$as$u20$core..convert..Into$LT$U$GT$$GT$::into::h7837bc8fb77e8181(self=<unavailable>) at mod.rs:726:9
    frame #11: 0x00000001060d8830 risingwave`risingwave_meta::error::MetaError::permission_denied::h6592a4f64415a283(s=<unavailable>) at error.rs:96:9
    frame #12: 0x00000001064bb43c risingwave`risingwave_meta::manager::catalog::CatalogManager$LT$S$GT$::drop_materialized_source::_$u7b$$u7b$closure$u7d$$u7d$::h36956825a3496ee1((null)=ResumeTy @ 0x0000000170c3e2b0) at mod.rs:1012:36
(... more callers omitted ...)

Heavy streaming tpch will cause high throughput (more than 600~700MB/s) to write to shared buffer while the sync operation would normally have a way less throughput to pump that out, causing shared buffer size to grow and swap. The example shows that the accumulation of shared buffer has a ~400MB/s input and ~200MB/s output, for 2 minutes, that would be 24G memory usage and cause severe page swap as well as database thrashing.

twocode@twocode-pc:/mnt/c/Users/dell$ sar -W 1 1000
21:46:12     pswpin/s pswpout/s
21:46:56         9.00  11329.00
21:46:57         0.00      0.00
21:46:58         0.00  45607.00
21:46:59         8.00      0.00
21:47:00         0.00   1024.00
21:47:01         7.00  28672.00
21:47:02         6.00  50476.00
21:47:03         3.00  67409.00
21:47:04        22.00  79243.00
21:47:05         4.00  45165.00

Is your feature request related to a problem? Please describe.

Implement create table as select statement to support duckdb test.

Describe the solution you'd like

No response

Describe alternatives you've considered

No response

Additional context

No response

For following sql:

select * from t order by a limit 10000, 10

And we assume that t is ordered by a in each patition, we should generate following plan:

   TopN(offset = 10000, limit = 10)
          |
    MergeSortExchange
            |
    TableScan(limit = 10010)

This way we don't need to pull the first 10000 rows into executor, which can avoid encoding/decoding of those rows. But the storage layer need to provide api for those skipping.

What's changed and what's your intention?

Implement Value using Executor2 trait.

Please explain IN DETAIL what the changes are in this PR and why they are needed:

• Summarize your change (mandatory)
• How does this PR work? Need a brief introduction for the changed logic (optional)
• Describe clearly one logical change and avoid lazy messages (optional)
• Describe any limitations of the current code (optional)

Checklist

• [x] I have written necessary docs and comments
• [x] I have added necessary unit tests and integration tests

Refer to a related PR or issue link (optional)

close https://github.com/singularity-data/risingwave/issues/1946

I hereby agree to the terms of the Singularity Data, Inc. Contributor License Agreement.

What's changed and what's your intention?

Enable batch system to evaluate NOW() as a function instead of an operator. result before this PR:

dev=> select now();
ERROR:  QueryError: internal error: Failed to build executor: Expr error: Unsupported function: Ok(Now)

result after this PR: test result:

dev=> select now();
          ?column?          
----------------------------
 2023-01-03 08:01:52.304273
(1 row)

dev=> create table t (v1 int);
CREATE_TABLE
dev=> create materialized view mv as select now() as nw from t where v1 >= 3;
ERROR:  QueryError: internal error: Rpc error: gRPC error (Internal error): Expression error: Expected session timestamp bound into Now
dev=> create table tt(v1 int, t1 timestamp); 
CREATE_TABLE
dev=> create materialized view mv as select * from tt where tt.t1 > now() - INTERVAL '1 hour';
CREATE_MATERIALIZED_VIEW

We insert a session timestamp into arguments of NOW(), which will only be used in batch system.

Checklist

• [x] I have written necessary rustdoc comments
• [x] All checks passed in ./risedev check (or alias, ./risedev c)

Refer to a related PR or issue link (optional)

#7139

I hereby agree to the terms of the Singularity Data, Inc. Contributor License Agreement.

What's changed and what's your intention?

Cancel compaction tasks and trigger new ones to its compaction group if no progress is made in num_blocks_completed or num_ssts_uploaded after a certain timeout.

Checklist

• [x] I have written necessary rustdoc comments
• [x] I have added necessary unit tests and integration tests
• [x] All checks passed in ./risedev check (or alias, ./risedev c)

Refer to a related PR or issue link (optional)

RFC: https://github.com/singularity-data/risingwave/issues/4457 Fixes: https://github.com/singularity-data/risingwave/issues/3677, https://github.com/singularity-data/risingwave/issues/4511, https://github.com/risingwavelabs/risingwave/issues/3677

Is your feature request related to a problem? Please describe. The extended query mode now already can run most of our original e2e test. We can add it in the ci.

Describe the solution you'd like

1. add e2e_test_extended in e2e_test folder, (e2e_test/e2e_test_extended)

   The e2e_test_extended is copy of e2e_test. The difference is: due to our extended query mode can't support {array, struct, interval, Decimal::Infinity} type now, I ignore some test case in e2e_test_extended by name these test case '.slt.part.un' instead of '.slt.part'. (When we support more type, we will reuse these test case. When e2e_test_extended support all test case, we can directly use e2e_test.)

2. add new CI script in ci/scripts

• e2e-source-test-extended.sh
• e2e-test-parallel-extended.sh
• e2e-test-extended.sh

I hereby agree to the terms of the Singularity Data, Inc. Contributor License Agreement.

What's changed and what's your intention?

This PR should implement CREATE SINK feature.

Please explain IN DETAIL what the changes are in this PR and why they are needed:

• Frontend now produces a StreamFragmentGraph which consists of an upstream ChainNode and a downstream SinkNode (src/frontend/src/handler/create_sink.rs).
• Frontend implements methods to add and delete sink to/from SchemaCatalog
• Meta receives the CreateSinkRequest from frontend. Dependent relations are checked + SinkManager is initialized.
• Via the SinkManager a create sink request is send to the Compute node.
• Compute node can add/remove Sink to/from MemSinkManager (located in src/sink repository).

./e2e_test/sink/basic_test.slt is runnable, however the only things that currently happens is that sink is added/removed to from the respective nodes and the communication between frontend/meta/compute. Basically the framework is there, but the actual implementation of create sink is still lacking.

Crucial missing functionalities:

• Meta: Actual actors need to be generated using ActorGraphBuilder and "connected" to the upstream materialized view.
• Compute: SinkExecutorBuilder needs to be called and executed.

Checklist

• [x] I have written necessary rustdoc comments
• [x] I have added necessary unit tests and integration tests
• [x] All checks passed in ./risedev check (or alias, ./risedev c)

Documentation

If your pull request contains user-facing changes, please specify the types of the changes, and create a release note. Otherwise, please feel free to remove this section.

Types of user-facing changes

Please keep the types that apply to your changes, and remove those that do not apply.

• Connector (sources & sinks)

Release note

Please create a release note for your changes. In the release note, focus on the impact on users, and mention the environment or conditions where the impact may occur.

Refer to a related PR or issue link (optional)

dev=> create table t1 (v1 int, v2 int);
CREATE_TABLE
dev=> create table t2 (v1 int, v2 int,);
CREATE_TABLE

dev=> create source s1 (v1 int, v2 int,) with ( connector = 'datagen') ROW FORMAT JSON;
CREATE_SOURCE
dev=> create source s1 (v1 int, v2 int,) with ( connector = 'datagen',) ROW FORMAT JSON;
ERROR:  QueryError: sql parser error: Expected identifier, found: )

Seems it does not hurt?

I hereby agree to the terms of the Singularity Data, Inc. Contributor License Agreement.

What's changed and what's your intention?

• Remove the create_table_inner and unify it into create_stream_job.
• Extract a visit_fragment utility and use it for filling source IDs / streaming job IDs.

Checklist

• [x] I have written necessary rustdoc comments
• [x] I have added necessary unit tests and integration tests
• [x] All checks passed in ./risedev check (or alias, ./risedev c)

Refer to a related PR or issue link (optional)

• #6903

I hereby agree to the terms of the Singularity Data, Inc. Contributor License Agreement.

What's changed and what's your intention?

As per the title, it will be used for Kafka performance testing setting as discussed with @huangjw806, who will use this to control how much time a query runs while all the query reads from the same kafka topic (one by one)

This is only used internally, so it will not go into the Doc.

Checklist

• [x] I have written necessary rustdoc comments
• [x] I have added necessary unit tests and integration tests
• [x] All checks passed in ./risedev check (or alias, ./risedev c)

I hereby agree to the terms of the Singularity Data, Inc. Contributor License Agreement.

What's changed and what's your intention?

Ensure actor task is completely dropped after calling drop_all_actors, see https://github.com/risingwavelabs/risingwave/issues/7208#issuecomment-1375060881. Also remove unused method wait_all.

Checklist

~~- [ ] I have written necessary rustdoc comments~~ ~~- [ ] I have added necessary unit tests and integration tests~~

• [x] All checks passed in ./risedev check (or alias, ./risedev c)

Documentation

If your pull request contains user-facing changes, please specify the types of the changes, and create a release note. Otherwise, please feel free to remove this section.

Types of user-facing changes

Please keep the types that apply to your changes, and remove those that do not apply.

• Installation and deployment
• Connector (sources & sinks)
• SQL commands, functions, and operators
• RisingWave cluster configuration changes
• Other (please specify in the release note below)

Release note

Please create a release note for your changes. In the release note, focus on the impact on users, and mention the environment or conditions where the impact may occur.

Refer to a related PR or issue link (optional)

fix #7208

I hereby agree to the terms of the Singularity Data, Inc. Contributor License Agreement.

What's changed and what's your intention?

As per the title

Checklist

• [x] I have written necessary rustdoc comments
• [x] I have added necessary unit tests and integration tests
• [x] All checks passed in ./risedev check (or alias, ./risedev c)

Refer to a related PR or issue link (optional)

I hereby agree to the terms of the Singularity Data, Inc. Contributor License Agreement.

What's changed and what's your intention?

WIP

This section will be used as the commit message. Please do not leave this empty!

Please explain IN DETAIL what the changes are in this PR and why they are needed:

• Summarize your change (mandatory)
• How does this PR work? Need a brief introduction for the changed logic (optional)
• Describe clearly one logical change and avoid lazy messages (optional)
• Describe any limitations of the current code (optional)

Checklist

• [ ] I have written necessary rustdoc comments
• [ ] I have added necessary unit tests and integration tests
• [ ] All checks passed in ./risedev check (or alias, ./risedev c)

Documentation

If your pull request contains user-facing changes, please specify the types of the changes, and create a release note. Otherwise, please feel free to remove this section.

Types of user-facing changes

Please keep the types that apply to your changes, and remove those that do not apply.

• Installation and deployment
• Connector (sources & sinks)
• SQL commands, functions, and operators
• RisingWave cluster configuration changes
• Other (please specify in the release note below)

Release note

Please create a release note for your changes. In the release note, focus on the impact on users, and mention the environment or conditions where the impact may occur.

Refer to a related PR or issue link (optional)