perf(connector): generate native Row when datagen to avoid json deser

[lmatz]

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

What's changed and what's your intention?

#4555 #4961

Use Box::into_raw to leak the memory, and then Box::from_raw to reclaim the ownership of that piece of memory.

introduce keywords NATIVE

create source ... with ( connector = 'datagen' ) ROW FORMAT NATIVE

and create a special native parser for that.

I guess creating individual rows and then sending them row by row is still suboptimal, compared to creating columns (column by column, so create, e.g. 1024, the same types of data each time) to compose a StreamChunk.

This is unsafe but limited to a small scope.
No reason to free the memory twice as it requires using the Bytes twice, which itself would suggest it's a suboptimal implementation.

Comparison on c5.4xLarge EC2.
Query:

create source t1 (
    v1 BIGINT, 
    v2 BIGINT, 
    t1 timestamp, 
    t2 timestamp,
    c1 varchar,
    c2 varchar
) with ( 
    connector = 'datagen', 
    datagen.split.num = '8',
    datagen.rows.per.second = '1000000000'
) ROW FORMAT JSON;

create sink s1 as select * from t1 with ( connector = 'blackhole' );

risedev.yml:

parallelism: 8

This pr:

commit: 94d508a:

So more than 2X improvement.

Guess the improvement can be artificially increased by specifying more and longer column names, and more and larger column values.

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)

Refer to a related PR or issue link (optional)

[codecov]

Codecov Report

Merging #6969 (5643393) into main (70e3132) will increase coverage by 72.88%.
The diff coverage is 86.84%.

@@            Coverage Diff            @@
##           main    #6969       +/-   ##
=========================================
+ Coverage      0   72.88%   +72.88%     
=========================================
  Files         0     1041     +1041     
  Lines         0   167213   +167213     
=========================================
+ Hits          0   121872   +121872     
- Misses        0    45341    +45341     

Flag	Coverage Δ
rust	72.88% <86.84%> (?)

Flags with carried forward coverage won't be shown. Click here to find out more.

Impacted Files	Coverage Δ
src/frontend/src/handler/create_source.rs	61.56% <0.00%> (ø)
src/source/src/lib.rs	81.81% <ø> (ø)
src/source/src/parser/csv_parser.rs	53.54% <0.00%> (ø)
src/source/src/parser/protobuf/parser.rs	54.16% <0.00%> (ø)
src/sqlparser/src/ast/statement.rs	69.79% <0.00%> (ø)
src/sqlparser/src/keywords.rs	100.00% <ø> (ø)
src/source/src/parser/canal/simd_json_parser.rs	62.80% <66.66%> (ø)
src/source/src/parser/maxwell/simd_json_parser.rs	64.28% <66.66%> (ø)
src/source/src/parser/native_parser.rs	96.00% <96.00%> (ø)
...c/connector/src/source/datagen/source/generator.rs	100.00% <100.00%> (ø)
... and 1049 more

📣 We’re building smart automated test selection to slash your CI/CD build times. Learn more

[lmatz]

#7032 will make this unnecessary, close now.