feat: dist sql analyze #1260

baojinri · 2023-10-13T04:36:21Z

Rationale

Currently, the analyze sql can not obtain detail metrics of partitioned table,

Detailed Changes

return metrics of partitioned table to remote client
collect metrics in client

Test Plan

Existing tests
add new integration tests for explain analyze

Rachelint · 2023-11-02T08:16:36Z

{
    "rows": [
        {
            "plan_type": "Plan with Metrics",
            "plan": "SortPreservingMergeExec: [minute@0 ASC NULLS LAST], metrics=[output_rows=6, elapsed_compute=46.541µs]
              SortExec: expr=[minute@0 ASC NULLS LAST], metrics=[output_rows=6, elapsed_compute=8ns, spill_count=0, spilled_bytes=0]
                ProjectionExec: expr=[time_bucket(cpu.timestamp,Utf8(\"PT60S\"))@0 as minute, MAX(cpu.usage_user)@1 as max_usage_user, MAX(cpu.usage_system)@2 as max_usage_system, MAX(cpu.usage_idle)@3 as max_usage_idle, MAX(cpu.usage_nice)@4 as max_usage_nice, MAX(cpu.usage_iowait)@5 as max_usage_iowait], metrics=[output_rows=6, elapsed_compute=35.333µs]
                  AggregateExec: mode=FinalPartitioned, gby=[time_bucket(cpu.timestamp,Utf8(\"PT60S\"))@0 as time_bucket(cpu.timestamp,Utf8(\"PT60S\"))], aggr=[MAX(cpu.usage_user), MAX(cpu.usage_system), MAX(cpu.usage_idle), MAX(cpu.usage_nice), MAX(cpu.usage_iowait)], metrics=[output_rows=6, elapsed_compute=479.835µs]
                    CoalesceBatchesExec: target_batch_size=8192, metrics=[output_rows=30, elapsed_compute=465.171µs]
                      RepartitionExec: partitioning=Hash([time_bucket(cpu.timestamp,Utf8(\"PT60S\"))@0], 8), input_partitions=3, metrics=[fetch_time=75.390247ms, repart_time=739.455µs, send_time=110.544µs]
                        ResolvedPartitionedScan: pushdown_continue:false, partition_count:3, metrics=[
                        cpu:
                __cpu_1:
                    poll_duration=1.2155ms
                    total_duration=24.775958ms
                    wait_duration=23.560458ms
                    CoalescePartitionsExec, metrics=[output_rows=12, elapsed_compute=27.292µs]
              AggregateExec: mode=Partial, gby=[time_bucket(timestamp@0, PT60S) as time_bucket(cpu.timestamp,Utf8(\"PT60S\"))], aggr=[MAX(cpu.usage_user), MAX(cpu.usage_system), MAX(cpu.usage_idle), MAX(cpu.usage_nice), MAX(cpu.usage_iowait)], metrics=[output_rows=12, elapsed_compute=555.584µs]
                RepartitionExec: partitioning=RoundRobinBatch(8), input_partitions=8, metrics=[fetch_time=16.261959ms, repart_time=8ns, send_time=29.715µs]
                  ProjectionExec: expr=[timestamp@0 as timestamp, usage_user@2 as usage_user, usage_system@3 as usage_system, usage_idle@4 as usage_idle, usage_nice@5 as usage_nice, usage_iowait@6 as usage_iowait], metrics=[output_rows=24, elapsed_compute=15.257µs]
                    ScanTable: table=__cpu_1, parallelism=8, metrics=[
                    :
                do_merge_sort=false
                chain_iter_0:
                    num_memtables=0
                    num_ssts=1
                    scan_duration=16.032959ms
                    since_create=16.576375ms
                    since_init=16.162834ms
                    total_batch_fetched=2
                    total_rows_fetched=24
                    scan_sst_1:
                        meta_data_cache_hit=true
                        parallelism=2
                        project_record_batch=17.160417ms
                        read_meta_data_duration=34.625µs
                        row_mem=1214000
                        row_num=14820
                        prune_row_groups:
                            pruned_by_custom_filter=0
                            pruned_by_min_max=0
                            row_groups_after_prune=2
                            total_row_groups=2
                            use_custom_filter=false
                            Predicate { exprs:[hostname IN ([Utf8(\"host_249\"), Utf8(\"host_1403\"), Utf8(\"host_1435\"), Utf8(\"host_3539\"), Utf8(\"host_3639\"), Utf8(\"host_3075\"), Utf8(\"host_815\"), Utf8(\"host_2121\")]), timestamp >= TimestampMillisecond(1662336000000, None), timestamp < TimestampMillisecond(1662339600000, None)], time_range:TimeRange { inclusive_start: Timestamp(1662336000000), exclusive_end: Timestamp(1662339600000) } }=0]
            :
                __cpu_3:
                    poll_duration=1.256419ms
                    total_duration=24.915875ms
                    wait_duration=23.659456ms
                    CoalescePartitionsExec, metrics=[output_rows=6, elapsed_compute=26.208µs]
              AggregateExec: mode=Partial, gby=[time_bucket(timestamp@0, PT60S) as time_bucket(cpu.timestamp,Utf8(\"PT60S\"))], aggr=[MAX(cpu.usage_user), MAX(cpu.usage_system), MAX(cpu.usage_idle), MAX(cpu.usage_nice), MAX(cpu.usage_iowait)], metrics=[output_rows=6, elapsed_compute=376.956µs]
                RepartitionExec: partitioning=RoundRobinBatch(8), input_partitions=8, metrics=[send_time=18.924µs, fetch_time=16.109503ms, repart_time=8ns]
                  ProjectionExec: expr=[timestamp@0 as timestamp, usage_user@2 as usage_user, usage_system@3 as usage_system, usage_idle@4 as usage_idle, usage_nice@5 as usage_nice, usage_iowait@6 as usage_iowait], metrics=[output_rows=6, elapsed_compute=9.423µs]
                    ScanTable: table=__cpu_3, parallelism=8, metrics=[
                    :
                do_merge_sort=false
                chain_iter_0:
                    num_memtables=0
                    num_ssts=1
                    scan_duration=15.926334ms
                    since_create=16.707083ms
                    since_init=16.000458ms
                    total_batch_fetched=1
                    total_rows_fetched=6
                    scan_sst_1:
                        meta_data_cache_hit=true
                        parallelism=2
                        project_record_batch=17.393667ms
                        read_meta_data_duration=3µs
                        row_mem=1214000
                        row_num=15042
                        prune_row_groups:
                            pruned_by_custom_filter=0
                            pruned_by_min_max=0
                            row_groups_after_prune=2
                            total_row_groups=2
                            use_custom_filter=false
                            Predicate { exprs:[hostname IN ([Utf8(\"host_249\"), Utf8(\"host_1403\"), Utf8(\"host_1435\"), Utf8(\"host_3539\"), Utf8(\"host_3639\"), Utf8(\"host_3075\"), Utf8(\"host_815\"), Utf8(\"host_2121\")]), timestamp >= TimestampMillisecond(1662336000000, None), timestamp < TimestampMillisecond(1662339600000, None)], time_range:TimeRange { inclusive_start: Timestamp(1662336000000), exclusive_end: Timestamp(1662339600000) } }=0]
            :
                __cpu_0:
                    poll_duration=1.163751ms
                    total_duration=24.882833ms
                    wait_duration=23.719082ms
                    CoalescePartitionsExec, metrics=[output_rows=12, elapsed_compute=30.333µs]
              AggregateExec: mode=Partial, gby=[time_bucket(timestamp@0, PT60S) as time_bucket(cpu.timestamp,Utf8(\"PT60S\"))], aggr=[MAX(cpu.usage_user), MAX(cpu.usage_system), MAX(cpu.usage_idle), MAX(cpu.usage_nice), MAX(cpu.usage_iowait)], metrics=[output_rows=12, elapsed_compute=658.501µs]
                RepartitionExec: partitioning=RoundRobinBatch(8), input_partitions=8, metrics=[send_time=26.465µs, fetch_time=16.535752ms, repart_time=8ns]
                  ProjectionExec: expr=[timestamp@0 as timestamp, usage_user@2 as usage_user, usage_system@3 as usage_system, usage_idle@4 as usage_idle, usage_nice@5 as usage_nice, usage_iowait@6 as usage_iowait], metrics=[output_rows=18, elapsed_compute=15.049µs]
                    ScanTable: table=__cpu_0, parallelism=8, metrics=[
                    :
                do_merge_sort=false
                chain_iter_0:
                    num_memtables=0
                    num_ssts=1
                    scan_duration=16.21ms
                    since_create=16.875791ms
                    since_init=16.325417ms
                    total_batch_fetched=2
                    total_rows_fetched=18
                    scan_sst_1:
                        meta_data_cache_hit=true
                        parallelism=2
                        project_record_batch=18.508416ms
                        read_meta_data_duration=40.583µs
                        row_mem=1214000
                        row_num=15228
                        prune_row_groups:
                            pruned_by_custom_filter=0
                            pruned_by_min_max=0
                            row_groups_after_prune=2
                            total_row_groups=2
                            use_custom_filter=false
                            Predicate { exprs:[hostname IN ([Utf8(\"host_249\"), Utf8(\"host_1403\"), Utf8(\"host_1435\"), Utf8(\"host_3539\"), Utf8(\"host_3639\"), Utf8(\"host_3075\"), Utf8(\"host_815\"), Utf8(\"host_2121\")]), timestamp >= TimestampMillisecond(1662336000000, None), timestamp < TimestampMillisecond(1662339600000, None)], time_range:TimeRange { inclusive_start: Timestamp(1662336000000), exclusive_end: Timestamp(1662339600000) } }=0]
            :
            =0]
                          AggregateExec: mode=Partial, gby=[time_bucket(timestamp@0, PT60S) as time_bucket(cpu.timestamp,Utf8(\"PT60S\"))], aggr=[MAX(cpu.usage_user), MAX(cpu.usage_system), MAX(cpu.usage_idle), MAX(cpu.usage_nice), MAX(cpu.usage_iowait)], metrics=[]
                            RepartitionExec: partitioning=RoundRobinBatch(8), input_partitions=8, metrics=[]
                              ProjectionExec: expr=[timestamp@0 as timestamp, usage_user@2 as usage_user, usage_system@3 as usage_system, usage_idle@4 as usage_idle, usage_nice@5 as usage_nice, usage_iowait@6 as usage_iowait], metrics=[]
                                UnresolvedSubTableScan: table:TableIdentifier { catalog: \"ceresdb\", schema: \"public\", table: \"__cpu_1\" }, table_scan_ctx:TableScanContext { read_parallelism: 8, batch_size: 8192, projection: Some([1, 2, 12, 13, 14, 15, 16]), predicate: Predicate { exprs:[hostname IN ([Utf8(\"host_249\"), Utf8(\"host_1403\"), Utf8(\"host_1435\"), Utf8(\"host_3539\"), Utf8(\"host_3639\"), Utf8(\"host_3075\"), Utf8(\"host_815\"), Utf8(\"host_2121\")]), timestamp >= TimestampMillisecond(1662336000000, None), timestamp < TimestampMillisecond(1662339600000, None)], time_range:TimeRange { inclusive_start: Timestamp(1662336000000), exclusive_end: Timestamp(1662339600000) } } }, partition_count:8, metrics=[]
                          AggregateExec: mode=Partial, gby=[time_bucket(timestamp@0, PT60S) as time_bucket(cpu.timestamp,Utf8(\"PT60S\"))], aggr=[MAX(cpu.usage_user), MAX(cpu.usage_system), MAX(cpu.usage_idle), MAX(cpu.usage_nice), MAX(cpu.usage_iowait)], metrics=[]
                            RepartitionExec: partitioning=RoundRobinBatch(8), input_partitions=8, metrics=[]
                              ProjectionExec: expr=[timestamp@0 as timestamp, usage_user@2 as usage_user, usage_system@3 as usage_system, usage_idle@4 as usage_idle, usage_nice@5 as usage_nice, usage_iowait@6 as usage_iowait], metrics=[]
                                UnresolvedSubTableScan: table:TableIdentifier { catalog: \"ceresdb\", schema: \"public\", table: \"__cpu_3\" }, table_scan_ctx:TableScanContext { read_parallelism: 8, batch_size: 8192, projection: Some([1, 2, 12, 13, 14, 15, 16]), predicate: Predicate { exprs:[hostname IN ([Utf8(\"host_249\"), Utf8(\"host_1403\"), Utf8(\"host_1435\"), Utf8(\"host_3539\"), Utf8(\"host_3639\"), Utf8(\"host_3075\"), Utf8(\"host_815\"), Utf8(\"host_2121\")]), timestamp >= TimestampMillisecond(1662336000000, None), timestamp < TimestampMillisecond(1662339600000, None)], time_range:TimeRange { inclusive_start: Timestamp(1662336000000), exclusive_end: Timestamp(1662339600000) } } }, partition_count:8, metrics=[]
                          AggregateExec: mode=Partial, gby=[time_bucket(timestamp@0, PT60S) as time_bucket(cpu.timestamp,Utf8(\"PT60S\"))], aggr=[MAX(cpu.usage_user), MAX(cpu.usage_system), MAX(cpu.usage_idle), MAX(cpu.usage_nice), MAX(cpu.usage_iowait)], metrics=[]
                            RepartitionExec: partitioning=RoundRobinBatch(8), input_partitions=8, metrics=[]
                              ProjectionExec: expr=[timestamp@0 as timestamp, usage_user@2 as usage_user, usage_system@3 as usage_system, usage_idle@4 as usage_idle, usage_nice@5 as usage_nice, usage_iowait@6 as usage_iowait], metrics=[]
                                UnresolvedSubTableScan: table:TableIdentifier { catalog: \"ceresdb\", schema: \"public\", table: \"__cpu_0\" }, table_scan_ctx:TableScanContext { read_parallelism: 8, batch_size: 8192, projection: Some([1, 2, 12, 13, 14, 15, 16]), predicate: Predicate { exprs:[hostname IN ([Utf8(\"host_249\"), Utf8(\"host_1403\"), Utf8(\"host_1435\"), Utf8(\"host_3539\"), Utf8(\"host_3639\"), Utf8(\"host_3075\"), Utf8(\"host_815\"), Utf8(\"host_2121\")]), timestamp >= TimestampMillisecond(1662336000000, None), timestamp < TimestampMillisecond(1662339600000, None)], time_range:TimeRange { inclusive_start: Timestamp(1662336000000), exclusive_end: Timestamp(1662339600000) } } }, partition_count:8, metrics=[]
                                "
        }
    ]
}

Above is the result I get in local test, do you notice the empty metric sub plans still exist? You can't just finish the feature in some hack ways, which will make codes not maintainable, and should think about how to impl it following the datafusion metrics framework.

AggregateExec: mode=Partial, gby=[time_bucket(timestamp@0, PT60S) as time_bucket(cpu.timestamp,Utf8(\"PT60S\"))], aggr=[MAX(cpu.usage_user), MAX(cpu.usage_system), MAX(cpu.usage_idle), MAX(cpu.usage_nice), MAX(cpu.usage_iowait)], metrics=[]

server/src/grpc/remote_engine_service/mod.rs

remote_engine_client/src/client.rs

query_engine/src/datafusion_impl/task_context.rs

query_engine/src/physical_planner.rs

remote_engine_client/src/client.rs

server/src/grpc/remote_engine_service/mod.rs

components/trace_metric/src/metric.rs

ShiKaiWi · 2023-11-15T01:57:16Z

@baojinri Could you provide an example of the analysis result?

baojinri · 2023-11-15T08:07:35Z

Example result:

SortPreservingMergeExec: [minute@0 ASC NULLS LAST], metrics=[output_rows=6, elapsed_compute=94.084µs]
  SortExec: expr=[minute@0 ASC NULLS LAST], metrics=[output_rows=6, elapsed_compute=8ns, spill_count=0, spilled_bytes=0]
    ProjectionExec: expr=[time_bucket(cpu.timestamp,Utf8("PT60S"))@0 as minute, MAX(cpu.usage_user)@1 as max_usage_user, MAX(cpu.usage_system)@2 as max_usage_system, MAX(cpu.usage_idle)@3 as max_usage_idle, MAX(cpu.usage_nice)@4 as max_usage_nice, MAX(cpu.usage_iowait)@5 as max_usage_iowait], metrics=[output_rows=6, elapsed_compute=30.916µs]
      AggregateExec: mode=FinalPartitioned, gby=[time_bucket(cpu.timestamp,Utf8("PT60S"))@0 as time_bucket(cpu.timestamp,Utf8("PT60S"))], aggr=[MAX(cpu.usage_user), MAX(cpu.usage_system), MAX(cpu.usage_idle), MAX(cpu.usage_nice), MAX(cpu.usage_iowait)], metrics=[output_rows=6, elapsed_compute=1.475875ms]
        CoalesceBatchesExec: target_batch_size=8192, metrics=[output_rows=30, elapsed_compute=432.305µs]
          RepartitionExec: partitioning=Hash([time_bucket(cpu.timestamp,Utf8("PT60S"))@0], 8), input_partitions=3, metrics=[repart_time=961.753µs, fetch_time=128.515041ms, send_time=69µs]
            ResolvedPartitionedScan: pushdown_continue:false, partition_count:3, metrics=[
cpu:
    __cpu_3:
        CoalescePartitionsExec, metrics=[output_rows=6, elapsed_compute=17.875µs]
  AggregateExec: mode=Partial, gby=[time_bucket(timestamp@0, PT60S) as time_bucket(cpu.timestamp,Utf8("PT60S"))], aggr=[MAX(cpu.usage_user), MAX(cpu.usage_system), MAX(cpu.usage_idle), MAX(cpu.usage_nice), MAX(cpu.usage_iowait)], metrics=[output_rows=6, elapsed_compute=471.709µs]
    RepartitionExec: partitioning=RoundRobinBatch(8), input_partitions=8, metrics=[send_time=7.799µs, repart_time=8ns, fetch_time=31.4ms]
      ProjectionExec: expr=[timestamp@0 as timestamp, usage_user@2 as usage_user, usage_system@3 as usage_system, usage_idle@4 as usage_idle, usage_nice@5 as usage_nice, usage_iowait@6 as usage_iowait], metrics=[output_rows=6, elapsed_compute=6.09µs]
        ScanTable: table=__cpu_3, parallelism=8, metrics=[
:
    do_merge_sort=false
    chain_iter_0:
        num_memtables=0
        num_ssts=1
        scan_duration=31.235834ms
        since_create=32.0695ms
        since_init=31.279791ms
        total_batch_fetched=1
        total_rows_fetched=6
        scan_sst_1:
            meta_data_cache_hit=true
            parallelism=2
            project_record_batch=51.54025ms
            read_meta_data_duration=72.792µs
            row_mem=1214000
            row_num=15042
            prune_row_groups:
                pruned_by_custom_filter=0
                pruned_by_min_max=0
                row_groups_after_prune=2
                total_row_groups=2
                use_custom_filter=false


Predicate { exprs:[hostname IN ([Utf8("host_249"), Utf8("host_1403"), Utf8("host_1435"), Utf8("host_3539"), Utf8("host_3639"), Utf8("host_3075"), Utf8("host_815"), Utf8("host_2121")]), timestamp >= TimestampMillisecond(1662336000000, None), timestamp < TimestampMillisecond(1662339600000, None)], time_range:TimeRange { inclusive_start: Timestamp(1662336000000), exclusive_end: Timestamp(1662339600000) } }=0]
=0
        poll_duration=1.374958ms
        total_duration=41.677834ms
        wait_duration=40.302876ms
    __cpu_0:
        CoalescePartitionsExec, metrics=[output_rows=12, elapsed_compute=27.25µs]
  AggregateExec: mode=Partial, gby=[time_bucket(timestamp@0, PT60S) as time_bucket(cpu.timestamp,Utf8("PT60S"))], aggr=[MAX(cpu.usage_user), MAX(cpu.usage_system), MAX(cpu.usage_idle), MAX(cpu.usage_nice), MAX(cpu.usage_iowait)], metrics=[output_rows=12, elapsed_compute=1.276168ms]
    RepartitionExec: partitioning=RoundRobinBatch(8), input_partitions=8, metrics=[repart_time=8ns, fetch_time=32.165252ms, send_time=14.132µs]
      ProjectionExec: expr=[timestamp@0 as timestamp, usage_user@2 as usage_user, usage_system@3 as usage_system, usage_idle@4 as usage_idle, usage_nice@5 as usage_nice, usage_iowait@6 as usage_iowait], metrics=[output_rows=18, elapsed_compute=7.257µs]
        ScanTable: table=__cpu_0, parallelism=8, metrics=[
:
    do_merge_sort=false
    chain_iter_0:
        num_memtables=0
        num_ssts=1
        scan_duration=32.040042ms
        since_create=32.657792ms
        since_init=32.101583ms
        total_batch_fetched=2
        total_rows_fetched=18
        scan_sst_1:
            meta_data_cache_hit=true
            parallelism=2
            project_record_batch=52.80775ms
            read_meta_data_duration=72µs
            row_mem=1214000
            row_num=15228
            prune_row_groups:
                pruned_by_custom_filter=0
                pruned_by_min_max=0
                row_groups_after_prune=2
                total_row_groups=2
                use_custom_filter=false


Predicate { exprs:[hostname IN ([Utf8("host_249"), Utf8("host_1403"), Utf8("host_1435"), Utf8("host_3539"), Utf8("host_3639"), Utf8("host_3075"), Utf8("host_815"), Utf8("host_2121")]), timestamp >= TimestampMillisecond(1662336000000, None), timestamp < TimestampMillisecond(1662339600000, None)], time_range:TimeRange { inclusive_start: Timestamp(1662336000000), exclusive_end: Timestamp(1662339600000) } }=0]
=0
        poll_duration=1.093877ms
        total_duration=42.014084ms
        wait_duration=40.920207ms
    __cpu_1:
        CoalescePartitionsExec, metrics=[output_rows=12, elapsed_compute=24.041µs]
  AggregateExec: mode=Partial, gby=[time_bucket(timestamp@0, PT60S) as time_bucket(cpu.timestamp,Utf8("PT60S"))], aggr=[MAX(cpu.usage_user), MAX(cpu.usage_system), MAX(cpu.usage_idle), MAX(cpu.usage_nice), MAX(cpu.usage_iowait)], metrics=[output_rows=12, elapsed_compute=1.471331ms]
    RepartitionExec: partitioning=RoundRobinBatch(8), input_partitions=8, metrics=[send_time=14.049µs, repart_time=8ns, fetch_time=29.984ms]
      ProjectionExec: expr=[timestamp@0 as timestamp, usage_user@2 as usage_user, usage_system@3 as usage_system, usage_idle@4 as usage_idle, usage_nice@5 as usage_nice, usage_iowait@6 as usage_iowait], metrics=[output_rows=24, elapsed_compute=11.84µs]
        ScanTable: table=__cpu_1, parallelism=8, metrics=[
:
    do_merge_sort=false
    chain_iter_0:
        num_memtables=0
        num_ssts=1
        scan_duration=29.849834ms
        since_create=30.8535ms
        since_init=29.925167ms
        total_batch_fetched=2
        total_rows_fetched=24
        scan_sst_1:
            meta_data_cache_hit=true
            parallelism=2
            project_record_batch=53.268082ms
            read_meta_data_duration=68.833µs
            row_mem=1214000
            row_num=14820
            prune_row_groups:
                pruned_by_custom_filter=0
                pruned_by_min_max=0
                row_groups_after_prune=2
                total_row_groups=2
                use_custom_filter=false


Predicate { exprs:[hostname IN ([Utf8("host_249"), Utf8("host_1403"), Utf8("host_1435"), Utf8("host_3539"), Utf8("host_3639"), Utf8("host_3075"), Utf8("host_815"), Utf8("host_2121")]), timestamp >= TimestampMillisecond(1662336000000, None), timestamp < TimestampMillisecond(1662339600000, None)], time_range:TimeRange { inclusive_start: Timestamp(1662336000000), exclusive_end: Timestamp(1662339600000) } }=0]
=0
        poll_duration=1.135876ms
        total_duration=40.919792ms
        wait_duration=39.783916ms
=0]

df_engine_extensions/src/dist_sql_query/resolver.rs

server/src/grpc/remote_engine_service/mod.rs

components/trace_metric/src/collector.rs

query_engine/src/datafusion_impl/task_context.rs

df_engine_extensions/src/dist_sql_query/resolver.rs

df_engine_extensions/src/dist_sql_query/physical_plan.rs

df_engine_extensions/src/dist_sql_query/mod.rs

df_engine_extensions/src/dist_sql_query/physical_plan.rs

remote_engine_client/Cargo.toml

server/src/grpc/remote_engine_service/mod.rs

Rachelint · 2023-12-08T03:45:47Z

Add an integration test case for it.

df_engine_extensions/src/dist_sql_query/physical_plan.rs

integration_tests/cases/env/cluster/ddl/partition_table.sql

df_engine_extensions/src/dist_sql_query/physical_plan.rs

Cargo.toml

Rachelint

LGTM

## Rationale In #1260, we implemented distributed analyze, but for query that are not analyze, metrics will be returned, which will lead to a decrease in query performance. Therefore, we will fix it in this PR, and metrics will not be returned for normal queries. ## Detailed Changes - Add is_analyze field to determine whether it is analyze ## Test Plan Existing tests --------- Co-authored-by: jiacai2050 <dev@liujiacai.net>

baojinri force-pushed the feat-dist-sql-analyze branch from fe189f3 to 9fafaca Compare October 30, 2023 05:44

baojinri marked this pull request as ready for review October 30, 2023 12:59

Rachelint reviewed Nov 2, 2023

View reviewed changes

ShiKaiWi reviewed Nov 7, 2023

View reviewed changes

Rachelint reviewed Nov 7, 2023

View reviewed changes

components/trace_metric/src/metric.rs Outdated Show resolved Hide resolved

baojinri force-pushed the feat-dist-sql-analyze branch from 7f5847e to cd5b6b5 Compare November 13, 2023 09:10

baojinri added 8 commits November 15, 2023 14:12

dist sql analyze

652ac5e

add handle metrics in dedup remote physical plan

0cb8c4c

modify code for review

ad74f03

modify code for review

6a372f6

use Metric::Number to store remote metrics

1417197

add metrics_collector in ExecutePlanRequest

8d617be

modify metrics_collector

8d2955f

avoid collect empty metrics of remote plan

a35f088

Rachelint reviewed Nov 15, 2023

View reviewed changes

use remote_metrics_collector to collect metrics

8051eec

baojinri force-pushed the feat-dist-sql-analyze branch from cd5b6b5 to 8051eec Compare November 16, 2023 04:26

modify name of remote metrics collector

a8e4672

Rachelint reviewed Nov 16, 2023

View reviewed changes

baojinri added 3 commits November 16, 2023 19:13

use Arc<Mutex<String>> to store remote metrics

ff1a457

use RemoteTaskContext in RemotePhysicalPlanExecutor trait

7d35d8b

set None to physical plan

99954b9

baojinri changed the base branch from main to dev December 5, 2023 11:38

baojinri force-pushed the feat-dist-sql-analyze branch 2 times, most recently from 726686d to 99954b9 Compare December 5, 2023 12:04

baojinri added 2 commits December 7, 2023 10:51

Merge branch 'dev' into feat-dist-sql-analyze

4b89185

Merge branch 'dev' into feat-dist-sql-analyze

cdfeda7

Rachelint reviewed Dec 8, 2023

View reviewed changes

baojinri added 3 commits December 8, 2023 14:10

fix

c1d56be

add integration test of explain analyze

6fa3aa0

add more integration tests for explain analyze

67a2192

Rachelint reviewed Dec 18, 2023

View reviewed changes

df_engine_extensions/src/dist_sql_query/physical_plan.rs Outdated Show resolved Hide resolved

Cargo.toml Outdated Show resolved Hide resolved

baojinri added 2 commits December 19, 2023 11:43

fix

4bb014e

Merge branch 'dev' into feat-dist-sql-analyze

fd6580c

baojinri force-pushed the feat-dist-sql-analyze branch from c838b10 to fd6580c Compare December 19, 2023 04:28

Rachelint approved these changes Dec 27, 2023

View reviewed changes

ShiKaiWi merged commit b02bac2 into apache:dev Dec 28, 2023
7 checks passed

baojinri mentioned this pull request Dec 28, 2023

refactor: avoid returning metrics in non-analyze sql #1410

Merged

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

feat: dist sql analyze #1260

feat: dist sql analyze #1260

baojinri commented Oct 13, 2023 •

edited

Rachelint commented Nov 2, 2023 •

edited

ShiKaiWi commented Nov 15, 2023

baojinri commented Nov 15, 2023

Rachelint commented Dec 8, 2023

Rachelint left a comment

feat: dist sql analyze #1260

feat: dist sql analyze #1260

Conversation

baojinri commented Oct 13, 2023 • edited

Rationale

Detailed Changes

Test Plan

Rachelint commented Nov 2, 2023 • edited

ShiKaiWi commented Nov 15, 2023

baojinri commented Nov 15, 2023

Rachelint commented Dec 8, 2023

Rachelint left a comment

Choose a reason for hiding this comment

baojinri commented Oct 13, 2023 •

edited

Rachelint commented Nov 2, 2023 •

edited