explainstatement相关测试

[zhuyanhuazhuyanhua]

1.
官网原文：- 同步查询接口允许用户提交查询，等待查询完成，并在一个 RPC 中获取查询结果。该接口适用于执行快速查询，例如 DDL、DCL 和简单的 DQL。但是，如果查询运行时间过长，可能会导致超时。因此，建议使用同步查询 API 来运行复杂的查询。
修改之后：+ 同步查询接口允许用户提交查询，等待查询完成，并在一个 RPC 中获取查询结果。该接口适用于执行快速查询，例如 DDL、DCL 和简单的 DQL。但是，如果查询运行时间过长，可能会导致超时。因此，建议使用异步查询 API 来运行复杂的查询。
在api\scdb_api.proto中也做了改动

[jingshi-ant]

看最后的测试结果是正常的？都pass了？
另外，如果执行了setup.sh（会重新设置公私钥），建议先 docker compose down 清理掉残留的mysql container，里面会有老的用户信息。

[zhuyanhuazhuyanhua]

2.测试

{
  "queries": [
    {
      "name": "explain percent select ",
      "query": "explain select tbl_1.plain_int_0,tbl_1.plain_datetime_0,tbl_1.plain_timestamp_0 from alice.tbl_1;",
      "mysql_query": ""
    }
  ]
}

2025-09-28 19:42:13.9287 INFO interpreter.go:219 subgraph of alice: nodes:{key:"0" value:{node_name:"runsql.0" op_type:"RunSQL" outputs:{key:"Out" value:{tensors:{name:"scdb.alice_tbl_1.plain_int_0.0" elem_type:INT64 option:REFERENCE annotation:{status:TENSORSTATUS_PRIVATE} ref_num:1} tensors:{name:"scdb.alice_tbl_1.plain_datetime_0.1" elem_type:DATETIME option:REFERENCE annotation:{status:TENSORSTATUS_PRIVATE} ref_num:1} tensors:{name:"scdb.alice_tbl_1.plain_timestamp_0.2" elem_type:TIMESTAMP option:REFERENCE annotation:{status:TENSORSTATUS_PRIVATE} ref_num:1}}} attributes:{key:"sql" value:{t:{name:".0" elem_type:STRING string_data:"select tbl_1.plain_int_0,tbl_1.plain_datetime_0,tbl_1.plain_timestamp_0 from alice.tbl_1;"}}} attributes:{key:"table_refs" value:{t:{name:".0" shape:{dim:{dim_value:1}} elem_type:STRING string_data:"alice.tbl_1"}}}}} nodes:{key:"1" value:{node_name:"publish.1" op_type:"Publish" inputs:{key:"In" value:{tensors:{name:"scdb.alice_tbl_1.plain_int_0.0" elem_type:INT64 option:REFERENCE annotation:{status:TENSORSTATUS_PRIVATE} ref_num:1} tensors:{name:"scdb.alice_tbl_1.plain_datetime_0.1" elem_type:DATETIME option:REFERENCE annotation:{status:TENSORSTATUS_PRIVATE} ref_num:1} tensors:{name:"scdb.alice_tbl_1.plain_timestamp_0.2" elem_type:TIMESTAMP option:REFERENCE annotation:{status:TENSORSTATUS_PRIVATE} ref_num:1}}} outputs:{key:"Out" value:{tensors:{name:"plain_int_0.3" elem_type:STRING annotation:{status:TENSORSTATUS_PRIVATE} string_data:"plain_int_0"} tensors:{name:"plain_datetime_0.4" elem_type:STRING annotation:{status:TENSORSTATUS_PRIVATE} string_data:"plain_datetime_0"} tensors:{name:"plain_timestamp_0.5" elem_type:STRING annotation:{status:TENSORSTATUS_PRIVATE} string_data:"plain_timestamp_0"}}}}} policy:{worker_num:1 pipelines:{subdags:{jobs:{node_ids:"0"}} subdags:{jobs:{node_ids:"1"}}}}

2025-09-28 19:42:13.9287 INFO submit_and_get_handler.go:83 |RequestID:|SessionID:2d28ec02-9c60-11f0-96b7-eaad763c7245|ActionName:SCDBSubmitAndGetHandler@/public/submit_and_get|CostTime:11.614892ms|Reason:|ErrorMsg:|Request:user:{user:{account_system_type:NATIVE_USER native_user:{name:"alice"}}} query:"explain select plain_int_0, plain_datetime_0, plain_timestamp_0 from alice_tbl_1;" db_name:"scdb"|ClientIP:172.18.0.1

2025-09-28 19:42:13.9287 INFO server.go:158 |GIN|status=200|method=POST|path=/public/submit_and_get|ip=172.18.0.1|latency=11.674445ms|

{
  "queries": [
    {
      "name": "rank ",
      "query": "explain select ta.plain_int_0,percent_rank() over (partition by ta.plain_int_0 order by ta.plain_float_0) as num from alice.tbl_0 as ta;",
      "mysql_query": ""
    }
  ]
}

2025-09-28 19:52:36.9287 INFO interpreter.go:219 subgraph of alice: nodes:{key:"0" value:{node_name:"runsql.0" op_type:"RunSQL" outputs:{key:"Out" value:{tensors:{name:"scdb.alice_tbl_0.plain_int_0.0" elem_type:INT64 option:REFERENCE annotation:{status:TENSORSTATUS_PRIVATE} ref_num:1} tensors:{name:"Column#124.1" elem_type:FLOAT64 option:REFERENCE annotation:{status:TENSORSTATUS_PRIVATE} ref_num:1}}} attributes:{key:"sql" value:{t:{name:".0" elem_type:STRING string_data:"select ta.plain_int_0,percent_rank() over (partition by ta.plain_int_0 order by ta.plain_float_0) as num from alice.tbl_0 as ta;"}}} attributes:{key:"table_refs" value:{t:{name:".0" shape:{dim:{dim_value:1}} elem_type:STRING string_data:"alice.tbl_0"}}}}} nodes:{key:"1" value:{node_name:"publish.1" op_type:"Publish" inputs:{key:"In" value:{tensors:{name:"scdb.alice_tbl_0.plain_int_0.0" elem_type:INT64 option:REFERENCE annotation:{status:TENSORSTATUS_PRIVATE} ref_num:1} tensors:{name:"Column#124.1" elem_type:FLOAT64 option:REFERENCE annotation:{status:TENSORSTATUS_PRIVATE} ref_num:1}}} outputs:{key:"Out" value:{tensors:{name:"plain_int_0.2" elem_type:STRING annotation:{status:TENSORSTATUS_PRIVATE} string_data:"plain_int_0"} tensors:{name:"num.3" elem_type:STRING annotation:{status:TENSORSTATUS_PRIVATE} string_data:"num"}}}}} policy:{worker_num:1 pipelines:{subdags:{jobs:{node_ids:"0"}} subdags:{jobs:{node_ids:"1"}}}}

2025-09-28 19:52:36.9287 INFO submit_and_get_handler.go:83 |RequestID:|SessionID:a05e3a2b-9c61-11f0-96b7-eaad763c7245|ActionName:SCDBSubmitAndGetHandler@/public/submit_and_get|CostTime:11.130669ms|Reason:|ErrorMsg:|Request:user:{user:{account_system_type:NATIVE_USER native_user:{name:"alice"}}} query:"explain select ta.plain_int_0, percent_rank() over(partition by ta.plain_int_0 order by ta.plain_float_0) as num from alice_tbl_0 as ta;" db_name:"scdb"|ClientIP:172.18.0.1

2025-09-28 19:52:36.9287 INFO server.go:158 |GIN|status=200|method=POST|path=/public/submit_and_get|ip=172.18.0.1|latency=11.16236ms|

{
  "queries": [
    {
      "name": "explain left  join",
      "query": "explain select ta.plain_int_0 from alice_tbl_1 as ta left join alice_tbl_2 as tb on ta.plain_int_0 = tb.plain_int_0;",
      "mysql_query": ""
    }
  ]
}

2025-09-28 19:56:05.9287 INFO interpreter.go:219 subgraph of alice: nodes:{key:"0" value:{node_name:"runsql.0" op_type:"RunSQL" outputs:{key:"Out" value:{tensors:{name:"scdb.alice_tbl_1.plain_int_0.0" elem_type:INT64 option:REFERENCE annotation:{status:TENSORSTATUS_PRIVATE} ref_num:1}}} attributes:{key:"sql" value:{t:{name:".0" elem_type:STRING string_data:"select ta.plain_int_0 from alice.tbl_1 as ta left join alice.tbl_2 as tb on ta.plain_int_0=tb.plain_int_0;"}}} attributes:{key:"table_refs" value:{t:{name:".0" shape:{dim:{dim_value:2}} elem_type:STRING string_data:"alice.tbl_1" string_data:"alice.tbl_2"}}}}} nodes:{key:"1" value:{node_name:"publish.1" op_type:"Publish" inputs:{key:"In" value:{tensors:{name:"scdb.alice_tbl_1.plain_int_0.0" elem_type:INT64 option:REFERENCE annotation:{status:TENSORSTATUS_PRIVATE} ref_num:1}}} outputs:{key:"Out" value:{tensors:{name:"plain_int_0.1" elem_type:STRING annotation:{status:TENSORSTATUS_PRIVATE} string_data:"plain_int_0"}}}}} policy:{worker_num:1 pipelines:{subdags:{jobs:{node_ids:"0"}} subdags:{jobs:{node_ids:"1"}}}}

2025-09-28 19:56:05.9287 INFO submit_and_get_handler.go:83 |RequestID:|SessionID:1cd65ed3-9c62-11f0-96b7-eaad763c7245|ActionName:SCDBSubmitAndGetHandler@/public/submit_and_get|CostTime:11.117878ms|Reason:|ErrorMsg:|Request:user:{user:{account_system_type:NATIVE_USER native_user:{name:"alice"}}} query:"explain select ta.plain_int_0 from alice_tbl_1 as ta left join alice_tbl_2 as tb on ta.plain_int_0 = tb.plain_int_0;" db_name:"scdb"|ClientIP:172.18.0.1

2025-09-28 19:56:05.9287 INFO server.go:158 |GIN|status=200|method=POST|path=/public/submit_and_get|ip=172.18.0.1|latency=11.165014ms|

[zhuyanhuazhuyanhua]

3.修改为OutColumns存放explain

// If parameter async is true, the query result will be notified by engine, this function will always return nil
func (app *App) runDQL(ctx context.Context, s *session, async bool) (*scql.SCDBQueryResultResponse, error) {
	compileReq, err := app.buildCompileRequest(ctx, s)
	if err != nil {
		return nil, err
	}
	intrpr := interpreter.NewInterpreter()
	compiledPlan, err := intrpr.Compile(ctx, compileReq)

	if err != nil {
		return nil, err
	}

	s.GetSessionVars().AffectedByGroupThreshold = compiledPlan.Warning.GetMayAffectedByGroupThreshold()
	s.GetSessionVars().GroupByThreshold = compileReq.CompileOpts.SecurityCompromise.GroupByThreshold
	logrus.Infof("Execution Plan:\n%s\n", compiledPlan.GetExplain().GetExeGraphDot())

	sessionStartParams := &scql.JobStartParams{
		JobId:         s.id,
		SpuRuntimeCfg: compiledPlan.GetSpuRuntimeConf(),
		TimeZone:      s.GetSessionVars().GetTimeZone(),
	}
	var partyCodes []string
	for _, p := range compiledPlan.Parties {
		partyCodes = append(partyCodes, p.GetCode())
	}

	var partyInfo *graph.PartyInfo
	{
		db := s.GetSessionVars().Storage
		var users []storage.User
		result := db.Model(&storage.User{}).Where("party_code in ?", partyCodes).Find(&users)
		if result.Error != nil {
			return nil, result.Error
		}
		partyMap := make(map[string]*graph.Participant)
		for _, u := range users {
			participant := &graph.Participant{
				PartyCode: u.PartyCode,
				Endpoints: strings.Split(u.EngineEndpoints, ";"),
				Token:     u.EngineToken,
				PubKey:    u.EnginePubKey,
			}
			partyMap[u.PartyCode] = participant
		}
		participants := make([]*graph.Participant, 0, len(partyCodes))
		for i, code := range partyCodes {
			party, exists := partyMap[code]
			if !exists {
				return nil, fmt.Errorf("could not find info for party %s", code)
			}
			participants = append(participants, party)
			sessionStartParams.Parties = append(sessionStartParams.Parties, &scql.JobStartParams_Party{
				Code:      code,
				Name:      code,
				Host:      party.Endpoints[0],
				Rank:      int32(i),
				PublicKey: party.PubKey,
			})
		}
		partyInfo = graph.NewPartyInfo(participants)
	}

	pbRequests := make(map[string]*scql.RunExecutionPlanRequest)
	for party, graph := range compiledPlan.SubGraphs {
		startParams, ok := proto.Clone(sessionStartParams).(*scql.JobStartParams)
		if !ok {
			return nil, fmt.Errorf("failed to clone session start params")
		}
		startParams.PartyCode = party
		cbURL := url.URL{
			Scheme: app.config.Protocol,
			Host:   app.config.SCDBHost,
			Path:   engineCallbackPath,
		}
		pbRequests[party] = &scql.RunExecutionPlanRequest{
			JobParams:     startParams,
			Graph:         graph,
			Async:         async,
			CallbackUrl:   cbURL.String(),
			GraphChecksum: &scql.GraphChecksum{CheckGraphChecksum: false},
		}
	}

	isExplain, err := isExplainQuery(s.request.GetQuery())
	if err != nil {
		return nil, err
	}

	if isExplain {
		explainTensor := &scql.Tensor{
			Name:     "explain_result",
			ElemType: scql.PrimitiveDataType_STRING,
			Option:   scql.TensorOptions_VALUE,
		}

		explainTensor.StringData = []string{compiledPlan.GetExplain().GetExeGraphDot()}

		return &scql.SCDBQueryResultResponse{
			Status: &scql.Status{
				Code:    int32(scql.Code_OK),
				Message: "ok", 
			},
			OutColumns:    []*scql.Tensor{explainTensor}, 
			ScdbSessionId: s.id,
			AffectedRows:  0,
			Warnings:      nil,
		}, nil
	}

	engineClient := executor.NewEngineClient(
		app.config.Engine.ClientMode,
		app.config.Engine.ClientTimeout,
		&bcfg.TLSConf{
			Mode:       app.config.Engine.TLSCfg.Mode,
			CertPath:   app.config.Engine.TLSCfg.CertFile,
			KeyPath:    app.config.Engine.TLSCfg.KeyFile,
			CACertPath: app.config.Engine.TLSCfg.CACertFile,
		},
		app.config.Engine.ContentType,
		app.config.Engine.Protocol,
	)
	s.engineStub = executor.NewEngineStub(
		s.id,
		app.config.Protocol,
		app.config.SCDBHost,
		engineCallbackPath,
		engineClient,
	)

	var outputNames []string
	for _, col := range compiledPlan.GetSchema().GetColumns() {
		outputNames = append(outputNames, col.GetName())
	}

	exec, err := executor.NewExecutor(pbRequests, outputNames, s.engineStub, s.id, partyInfo)
	if err != nil {
		return nil, err
	}
	s.executor = exec
	resp, err := exec.RunExecutionPlan(ctx, async)
	if err != nil {
		return nil, err
	}

	if async {
		// In async mode, result will be set in callback
		return nil, nil
	}
	return resp, nil
}

[zhuyanhuazhuyanhua]

https://github.com/secretflow/scql/pull/675

[zhuyanhuazhuyanhua]

4. 详见 https://blog.csdn.net/2301_79815382/article/details/151257417?fromshare=blogdetail&sharetype=blogdetail&sharerId=151257417&sharerefer=PC&sharesource=2301_79815382&sharefrom=from_link
   runSQL函数核心处理在run函数中，run函数主要处理DDL/DCL：

// Run runs an DDL/DCL statement on SCDB
func Run(ctx sessionctx.Context, stmt string, is infoschema.InfoSchema) ([]*scql.Tensor, error) {
	// Step 1: Parsing
	p := parser.New()
	ast, err := p.ParseOneStmt(stmt, "", "")
	if err != nil {
		return nil, err
	}
	if err := core.Preprocess(ctx, ast, is); err != nil {
		return nil, err
	}

	// Step 2: Planning
	lp, _, err := core.BuildLogicalPlan(context.Background(), ctx, ast, is)
	if err != nil {
		return nil, err
	}

	// Step 3: Executing
	eb := newExecutorBuilder(ctx, is)
	exec := eb.build(lp)
	if eb.err != nil {
		return nil, eb.err
	}
	if err := exec.Open(context.Background()); err != nil {
		return nil, err
	}

	retTypes := []*types.FieldType{}
	for _, c := range lp.Schema().Columns {
		retTypes = append(retTypes, c.RetType)
	}

	ck := chunk.New(retTypes, ResultMaxRows, ResultMaxRows)
	var result []*scql.Tensor
	for {
		if err := exec.Next(context.Background(), ck); err != nil {
			return nil, err
		}
		if result, err = mergeResultFromChunk(ck, lp, result); err != nil {
			return nil, err
		}
		if ck.NumRows() == 0 {
			break
		}
	}
	return result, err
}

runDQL函数处理DQL：

// If parameter async is true, the query result will be notified by engine, this function will always return nil
func (app *App) runDQL(ctx context.Context, s *session, async bool) (*scql.SCDBQueryResultResponse, error) {
	compileReq, err := app.buildCompileRequest(ctx, s)
	if err != nil {
		return nil, err
	}
	intrpr := interpreter.NewInterpreter()
	compiledPlan, err := intrpr.Compile(ctx, compileReq)

	if err != nil {
		return nil, err
	}

	s.GetSessionVars().AffectedByGroupThreshold = compiledPlan.Warning.GetMayAffectedByGroupThreshold()
	s.GetSessionVars().GroupByThreshold = compileReq.CompileOpts.SecurityCompromise.GroupByThreshold
	logrus.Infof("Execution Plan:\n%s\n", compiledPlan.GetExplain().GetExeGraphDot())

	sessionStartParams := &scql.JobStartParams{
		JobId:         s.id,
		SpuRuntimeCfg: compiledPlan.GetSpuRuntimeConf(),
		TimeZone:      s.GetSessionVars().GetTimeZone(),
	}
	var partyCodes []string
	for _, p := range compiledPlan.Parties {
		partyCodes = append(partyCodes, p.GetCode())
	}

	var partyInfo *graph.PartyInfo
	{
		db := s.GetSessionVars().Storage
		var users []storage.User
		result := db.Model(&storage.User{}).Where("party_code in ?", partyCodes).Find(&users)
		if result.Error != nil {
			return nil, result.Error
		}
		partyMap := make(map[string]*graph.Participant)
		for _, u := range users {
			participant := &graph.Participant{
				PartyCode: u.PartyCode,
				Endpoints: strings.Split(u.EngineEndpoints, ";"),
				Token:     u.EngineToken,
				PubKey:    u.EnginePubKey,
			}
			partyMap[u.PartyCode] = participant
		}
		participants := make([]*graph.Participant, 0, len(partyCodes))
		for i, code := range partyCodes {
			party, exists := partyMap[code]
			if !exists {
				return nil, fmt.Errorf("could not find info for party %s", code)
			}
			participants = append(participants, party)
			sessionStartParams.Parties = append(sessionStartParams.Parties, &scql.JobStartParams_Party{
				Code:      code,
				Name:      code,
				Host:      party.Endpoints[0],
				Rank:      int32(i),
				PublicKey: party.PubKey,
			})
		}
		partyInfo = graph.NewPartyInfo(participants)
	}

	pbRequests := make(map[string]*scql.RunExecutionPlanRequest)
	for party, graph := range compiledPlan.SubGraphs {
		startParams, ok := proto.Clone(sessionStartParams).(*scql.JobStartParams)
		if !ok {
			return nil, fmt.Errorf("failed to clone session start params")
		}
		startParams.PartyCode = party
		cbURL := url.URL{
			Scheme: app.config.Protocol,
			Host:   app.config.SCDBHost,
			Path:   engineCallbackPath,
		}
		pbRequests[party] = &scql.RunExecutionPlanRequest{
			JobParams:     startParams,
			Graph:         graph,
			Async:         async,
			CallbackUrl:   cbURL.String(),
			GraphChecksum: &scql.GraphChecksum{CheckGraphChecksum: false},
		}
	}

	isExplain, err := isExplainQuery(s.request.GetQuery())
	if err != nil {
		return nil, err
	}

	if isExplain {
		explainTensor := &scql.Tensor{
			Name:     "explain_result",
			ElemType: scql.PrimitiveDataType_STRING,
			Option:   scql.TensorOptions_VALUE,
		}

		explainTensor.StringData = []string{compiledPlan.GetExplain().GetExeGraphDot()}

		return &scql.SCDBQueryResultResponse{
			Status: &scql.Status{
				Code:    int32(scql.Code_OK),
				Message: "ok",
			},
			OutColumns:    []*scql.Tensor{explainTensor},
			ScdbSessionId: s.id,
			AffectedRows:  0,
			Warnings:      nil,
		}, nil
	}

	engineClient := executor.NewEngineClient(
		app.config.Engine.ClientMode,
		app.config.Engine.ClientTimeout,
		&bcfg.TLSConf{
			Mode:       app.config.Engine.TLSCfg.Mode,
			CertPath:   app.config.Engine.TLSCfg.CertFile,
			KeyPath:    app.config.Engine.TLSCfg.KeyFile,
			CACertPath: app.config.Engine.TLSCfg.CACertFile,
		},
		app.config.Engine.ContentType,
		app.config.Engine.Protocol,
	)
	s.engineStub = executor.NewEngineStub(
		s.id,
		app.config.Protocol,
		app.config.SCDBHost,
		engineCallbackPath,
		engineClient,
	)

	var outputNames []string
	for _, col := range compiledPlan.GetSchema().GetColumns() {
		outputNames = append(outputNames, col.GetName())
	}

	exec, err := executor.NewExecutor(pbRequests, outputNames, s.engineStub, s.id, partyInfo)
	if err != nil {
		return nil, err
	}
	s.executor = exec
	resp, err := exec.RunExecutionPlan(ctx, async)
	if err != nil {
		return nil, err
	}

	if async {
		// In async mode, result will be set in callback
		return nil, nil
	}
	return resp, nil
}

run函数中不存在编译计划图过程，而runDQL函数有compiledPlan, err := intrpr.Compile(ctx, compileReq)以及compiledPlan.GetExplain().GetExeGraphDot()，存在编译计划图过程；正是explain所需的。