[BUG] Resources exhausted during export data.

[Benxiaohai001]

Describe the bug
A clear and concise description of what the bug is.
Resources exhausted during export data.
pre:
version:05d0e4da412cdd1474258774c71bebe91fe76bce
1meta 2data
meta：nohup ./target/release/cnosdb-meta --config ./config/config_21001.toml > /home/cnosdb/logs/meta_node.1.log
data：./target/release/cnosdb run -c 64 -m 200 --config ./config/config_31001.toml
settings：
max_buffer_size=256
others default
cat /nvme/data/cnodb-iot-seed-123-scale-100-2020.gz | gunzip | load_cnosdb --batch-size 30000 --reporting-period '10s' --urls http://192.168.0.24:31001 --workers 200
To Reproduce
Steps to reproduce the behavior:
copy into 'file:///data/parquet/readings' from readings file_format=(type='parquet');

Expected behavior
A clear and concise description of what you expected to happen.
no error

Additional context
Add any other context about the problem here.
"422 Unprocessable Entity, details: {"error_code":"010001","error_message":"Datafusion: External error: Resources exhausted: Failed to allocate additional 34542560 bytes for RepartitionExec[45] with 0 bytes already allocated - maximum available is 29578128"}"

[Benxiaohai001]

version :6c5b09f79bf96d5b98ae55557842dd3af78f8990
exec plan :

`test ❯ explain copy into 'file:///data/parquet/readings' from readings file_format=(type='parquet');
+---------------+----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
| plan_type     | plan                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                       |
+---------------+----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
| logical_plan  | Aggregate: groupBy=[[]], aggr=[[SUM(rows) AS rows]]                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                        |
|               |   TableWriter: not_exec_only_prevent_col_prune(CAST(time AS Utf8), device_version, driver, fleet, model, name, CAST(elevation AS Utf8), CAST(fuel_capacity AS Utf8), CAST(fuel_consumption AS Utf8), CAST(grade AS Utf8), CAST(heading AS Utf8), CAST(latitude AS Utf8), CAST(load_capacity AS Utf8), CAST(longitude AS Utf8), CAST(nominal_fuel_consumption AS Utf8), CAST(velocity AS Utf8)) AS rows                                                                                                                                                                     |
|               |     time := Timestamp(Nanosecond, None)                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                    |
|               |     device_version := Utf8                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                 |
|               |     driver := Utf8                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                         |
|               |     fleet := Utf8                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                          |
|               |     model := Utf8                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                          |
|               |     name := Utf8                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                           |
|               |     elevation := Float64                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                   |
|               |     fuel_capacity := Float64                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                               |
|               |     fuel_consumption := Float64                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                            |
|               |     grade := Float64                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                       |
|               |     heading := Float64                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                     |
|               |     latitude := Float64                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                    |
|               |     load_capacity := Float64                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                               |
|               |     longitude := Float64                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                   |
|               |     nominal_fuel_consumption := Float64                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                    |
|               |     velocity := Float64                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                    |
|               |     Projection: readings.time AS time, readings.device_version AS device_version, readings.driver AS driver, readings.fleet AS fleet, readings.model AS model, readings.name AS name, readings.elevation AS elevation, readings.fuel_capacity AS fuel_capacity, readings.fuel_consumption AS fuel_consumption, readings.grade AS grade, readings.heading AS heading, readings.latitude AS latitude, readings.load_capacity AS load_capacity, readings.longitude AS longitude, readings.nominal_fuel_consumption AS nominal_fuel_consumption, readings.velocity AS velocity |
|               |       TableScan: readings projection=[time, device_version, driver, fleet, model, name, elevation, fuel_capacity, fuel_consumption, grade, heading, latitude, load_capacity, longitude, nominal_fuel_consumption, velocity]                                                                                                                                                                                                                                                                                                                                                |
| physical_plan | AggregateExec: mode=Final, gby=[], aggr=[rows]                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                             |
|               |   CoalescePartitionsExec                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                   |
|               |     AggregateExec: mode=Partial, gby=[], aggr=[rows]                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                       |
|               |       TableWriterExec: [file:///data/parquet/readings]                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                     |
|               |         ProjectionExec: expr=[time@0 as time, device_version@1 as device_version, driver@2 as driver, fleet@3 as fleet, model@4 as model, name@5 as name, elevation@6 as elevation, fuel_capacity@7 as fuel_capacity, fuel_consumption@8 as fuel_consumption, grade@9 as grade, heading@10 as heading, latitude@11 as latitude, load_capacity@12 as load_capacity, longitude@13 as longitude, nominal_fuel_consumption@14 as nominal_fuel_consumption, velocity@15 as velocity]                                                                                            |
|               |           TskvExec: limit=None, predicate=ColumnDomains { column_to_domain: Some({}) }, projection=[time,device_version,driver,fleet,model,name,elevation,fuel_capacity,fuel_consumption,grade,heading,latitude,load_capacity,longitude,nominal_fuel_consumption,velocity]                                                                                                                                                                                                                                                                                                 |
|               |                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                            |
+---------------+----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
Query took 0.020 seconds.
`
`test ❯ copy into 'file:///data/parquet/readings' from readings file_format=(type='parquet');
"422 Unprocessable Entity, details: {\"error_code\":\"010001\",\"error_message\":\"Datafusion: Resources exhausted: Failed to allocate additional 35050464 bytes for TableScanMetrics[30] with 8582192960 bytes already allocated - maximum available is 15862872\"}"`

[Benxiaohai001]

The bug was fixed,So close the issue.
version: 4b16931

benchmark ❯ copy into 'file:///data/parquet/readings' from readings file_format=(type='parquet');
+-----------+
| rows      |
+-----------+
| 259743634 |
+-----------+
Query took 851.348 seconds.
[root@cicd_ujv23 cnosdb]# ll /data/parquet/readings/ -h
total 9.0G
-rw-r--r--. 1 root root  4.7M Jun 26 14:21 part-11593008481671005477-52-0.parquet
-rw-r--r--. 1 root root  1.2M Jun 26 14:32 part-11593008481671005477-52-100.parquet
-rw-r--r--. 1 root root  1.2M Jun 26 14:33 part-11593008481671005477-52-101.parquet
-rw-r--r--. 1 root root  1.2M Jun 26 14:33 part-11593008481671005477-52-102.parquet
-rw-r--r--. 1 root root  1.2M Jun 26 14:33 part-11593008481671005477-52-103.parquet
-rw-r--r--. 1 root root  1.2M Jun 26 14:33 part-11593008481671005477-52-104.parquet
...