Adjust Stmt duration as sum value and format aggr statemnt format (ma…

…trixorigin#11625) 1. improve aggr stmt‘s duration as aggrated sum value. 2. format aggr stmt's statement info format 1. with `/* #num queries */` prefix by default; 2. connect diff stmt with `";\n"` string. Approved by: @gavinyue, @heni02
fengttt · Sep 5, 2023 · a35a9fc · a35a9fc
1 parent 80f0e8b
commit a35a9fc
Show file tree

Hide file tree

Showing 5 changed files with 20 additions and 20 deletions.
diff --git a/pkg/util/trace/impl/motrace/Aggr_test.go b/pkg/util/trace/impl/motrace/Aggr_test.go
@@ -187,24 +187,24 @@ func TestAggregator(t *testing.T) {
 	if len(results) != 4 {
 		t.Errorf("Expected 4 aggregated statements, got %d", len(results))
 	}
-	assert.Equal(t, aggrWindow, results[0].(*StatementInfo).Duration)
-	assert.Equal(t, aggrWindow, results[1].(*StatementInfo).Duration)
-	assert.Equal(t, aggrWindow, results[2].(*StatementInfo).Duration)
-	assert.Equal(t, aggrWindow, results[3].(*StatementInfo).Duration)
+	assert.Equal(t, 50*time.Millisecond, results[0].(*StatementInfo).Duration)
+	assert.Equal(t, 50*time.Millisecond, results[1].(*StatementInfo).Duration)
+	assert.Equal(t, 50*time.Millisecond, results[2].(*StatementInfo).Duration)
+	assert.Equal(t, 50*time.Millisecond, results[3].(*StatementInfo).Duration)
 	require.Equal(t, []byte(`[2,5,10.000,15,20,25]`), results[0].(*StatementInfo).ExecPlan2Stats(ctx))
 	require.Equal(t, []byte(`[2,5,10.000,15,20,25]`), results[1].(*StatementInfo).ExecPlan2Stats(ctx))
 	require.Equal(t, []byte(`[2,5,10.000,15,20,25]`), results[2].(*StatementInfo).ExecPlan2Stats(ctx))
 	require.Equal(t, []byte(`[2,5,10.000,15,20,25]`), results[3].(*StatementInfo).ExecPlan2Stats(ctx))
 	item, _ := results[0].(*StatementInfo)
 	row := item.GetTable().GetRow(ctx)
 	results[0].(*StatementInfo).FillRow(ctx, row)
-	require.Equal(t, []byte(`[2,5,0.020,15,20,25]`), results[0].(*StatementInfo).ExecPlan2Stats(ctx))
+	require.Equal(t, []byte(`[2,5,2.000,15,20,25]`), results[0].(*StatementInfo).ExecPlan2Stats(ctx))
 	results[1].(*StatementInfo).FillRow(ctx, row)
-	require.Equal(t, []byte(`[2,5,0.020,15,20,25]`), results[1].(*StatementInfo).ExecPlan2Stats(ctx))
+	require.Equal(t, []byte(`[2,5,2.000,15,20,25]`), results[1].(*StatementInfo).ExecPlan2Stats(ctx))
 	results[2].(*StatementInfo).FillRow(ctx, row)
-	require.Equal(t, []byte(`[2,5,0.020,15,20,25]`), results[2].(*StatementInfo).ExecPlan2Stats(ctx))
+	require.Equal(t, []byte(`[2,5,2.000,15,20,25]`), results[2].(*StatementInfo).ExecPlan2Stats(ctx))
 	results[3].(*StatementInfo).FillRow(ctx, row)
-	require.Equal(t, []byte(`[2,5,0.020,15,20,25]`), results[3].(*StatementInfo).ExecPlan2Stats(ctx))
+	require.Equal(t, []byte(`[2,5,2.000,15,20,25]`), results[3].(*StatementInfo).ExecPlan2Stats(ctx))
 
 	aggregator.Close()
 
@@ -264,14 +264,14 @@ func TestAggregator(t *testing.T) {
 	assert.Equal(t, "Update 11", results[0].(*StatementInfo).StmtBuilder.String())
 	// should have two results since they have different sqlSourceType
 	assert.Equal(t, "Update 11", results[1].(*StatementInfo).StmtBuilder.String())
-	assert.Equal(t, aggrWindow, results[1].(*StatementInfo).Duration)
+	assert.Equal(t, 60*time.Millisecond, results[1].(*StatementInfo).Duration)
 	// RequestAt should be starting of the window
 	assert.Equal(t, fixedTime.Add(4*time.Second), results[0].(*StatementInfo).RequestAt)
 	// ResponseAt should be end of the window
 	assert.Equal(t, fixedTime.Add(9*time.Second), results[0].(*StatementInfo).ResponseAt)
 	require.Equal(t, []byte(`[2,5,10.000,15,20,25]`), results[0].(*StatementInfo).ExecPlan2Stats(ctx))
 	results[0].(*StatementInfo).FillRow(ctx, row)
-	require.Equal(t, []byte(`[2,5,0.024,15,20,25]`), results[0].(*StatementInfo).ExecPlan2Stats(ctx))
+	require.Equal(t, []byte(`[2,5,2.000,15,20,25]`), results[0].(*StatementInfo).ExecPlan2Stats(ctx))
 
 	_, err = aggregator.AddItem(&StatementInfo{
 		Account:       "MO",
@@ -347,11 +347,11 @@ func TestAggregatorWithStmtMerge(t *testing.T) {
 		t.Errorf("Expected 0 aggregated statements, got %d", len(results))
 	}
 
-	assert.Equal(t, "SELECT 11\nSELECT 11", results[0].(*StatementInfo).StmtBuilder.String())
+	assert.Equal(t, "SELECT 11;\nSELECT 11", results[0].(*StatementInfo).StmtBuilder.String())
 
-	res := "/*" + strconv.FormatInt(results[0].(*StatementInfo).AggrCount, 10) + " queries */ \n" + results[0].(*StatementInfo).StmtBuilder.String()
+	res := "/* " + strconv.FormatInt(results[0].(*StatementInfo).AggrCount, 10) + " queries */ \n" + results[0].(*StatementInfo).StmtBuilder.String()
 
-	assert.Equal(t, "/*2 queries */ \nSELECT 11\nSELECT 11", res)
+	assert.Equal(t, "/* 2 queries */ \nSELECT 11;\nSELECT 11", res)
 
 	assert.Equal(t, int64(2), results[0].(*StatementInfo).RowsRead)
 

diff --git a/pkg/util/trace/impl/motrace/buffer_pipe.go b/pkg/util/trace/impl/motrace/buffer_pipe.go
@@ -191,8 +191,8 @@ func genETLData(ctx context.Context, in []IBuffer2SqlItem, buf *bytes.Buffer, fa
 
 			stmt, stmt_ok := i.(*StatementInfo)
 			if stmt_ok {
-				if GetTracerProvider().enableStmtMerge {
-					stmt.Statement = "/*" + strconv.FormatInt(stmt.AggrCount, 10) + " queries */ \n" + stmt.StmtBuilder.String()
+				if stmt.AggrCount > 0 {
+					stmt.Statement = "/* " + strconv.FormatInt(stmt.AggrCount, 10) + " queries */ \n" + stmt.StmtBuilder.String()
 				} else {
 					stmt.Statement = stmt.StmtBuilder.String()
 				}

diff --git a/pkg/util/trace/impl/motrace/report_statement.go b/pkg/util/trace/impl/motrace/report_statement.go
@@ -78,7 +78,6 @@ func StatementInfoNew(i Item, ctx context.Context) Item {
 		s.Database = ""
 		s.StmtBuilder.WriteString(s.Statement)
 		duration := s.Duration
-		s.Duration = windowSize
 		s.AggrMemoryTime = mustDecimal128(convertFloat64ToDecimal128(s.statsArray.GetMemorySize() * float64(duration)))
 		s.RequestAt = s.ResponseAt.Truncate(windowSize)
 		s.ResponseAt = s.RequestAt.Add(windowSize)
@@ -93,10 +92,11 @@ func StatementInfoUpdate(existing, new Item) {
 	n := new.(*StatementInfo)
 	// update the stats
 	if GetTracerProvider().enableStmtMerge {
-		e.StmtBuilder.WriteString("\n")
+		e.StmtBuilder.WriteString(";\n")
 		e.StmtBuilder.WriteString(n.Statement)
 	}
 	e.AggrCount += 1
+	e.Duration += n.Duration
 	e.RowsRead += n.RowsRead
 	e.BytesScan += n.BytesScan
 	e.ResultCount += n.ResultCount

diff --git a/test/distributed/cases/log/query_tae_table.result b/test/distributed/cases/log/query_tae_table.result
@@ -35,9 +35,9 @@ create user u identified by '******'
 create user if not exists abc1 identified by '******', abc2 identified by '******', abc3 identified by '******', abc3 identified by '******'
 create external table t (a int) URL s3option{'endpoint'='s3.us-west-2.amazonaws.com', 'access_key_id'='******', 'secret_access_key'='******', 'bucket'='test', 'filepath'='*.txt', 'region'='us-west-2'}
 select 1
-select account, statement from system.statement_info where statement = 'insert into query_tae_table values (1)' and statement_id = @uuid_insert_table limit 1;
+select account, statement from system.statement_info where statement_id = @uuid_insert_table limit 1;
 account    statement
-query_tae_table    insert into query_tae_table values (1)
+query_tae_table    /* 1 queries */ \ninsert into query_tae_table values (1)
 select IF(span_kind="internal", 1, IF(span_kind="statement", 1, IF(span_kind="session", 1, IF(span_kind="remote", 1, 0)))) as exist from system.rawlog where `raw_item` = "log_info" limit 1;
 exist
 1

diff --git a/test/distributed/cases/log/query_tae_table.sql b/test/distributed/cases/log/query_tae_table.sql
@@ -34,7 +34,7 @@ select account from system.statement_info where statement_id = @uuid_create_tabl
 -- hide result
 select statement from system.statement_info where statement_id in (@uuid_hide_1, @uuid_hide_2, @uuid_hide_3, @uuid_hide_4);
 
-select account, statement from system.statement_info where statement = 'insert into query_tae_table values (1)' and statement_id = @uuid_insert_table limit 1;
+select account, statement from system.statement_info where statement_id = @uuid_insert_table limit 1;
 -- @session
 
 -- case: select span_kind issue #7571