Functionality for noRandomDuplicates parameter and custom Stream options with seeding

cmd/round/main.go

@@ -184,7 +184,7 @@ func processRoundDecimalFile(inputPath string) (err error) {
 				if err != nil {
 					return err
 				}
-				bufWriter = bufio.NewWriter(outputFile)
+				bufWriter = bufio.NewWriterSize(outputFile, 8192)
 				colCount = len(cols)
 			} else {
 				// no decimal column, quick exit

stage/map.go

@@ -108,6 +108,13 @@ func ParseStage(stage *Stage, stages Map) (*Stage, error) {
 		}
 	}
 	stages[stage.Id] = stage
+
+	// Set the seed for this stage (only relevant for non-stream stages)
+	// Stream stages will have their seeds set during runAsMultipleStreams
+	if stage.StreamCount == nil || *stage.StreamCount <= 1 {
+		stage.seed = stage.States.RandSeed
+	}
+
 	for _, nextStagePath := range stage.NextStagePaths {
 		if nextStage, err := ParseStageFromFile(nextStagePath, stages); err != nil {
 			return nil, err

stage/mysql_run_recorder.go

@@ -4,9 +4,10 @@ import (
 	"context"
 	"database/sql"
 	_ "embed"
-	_ "github.com/go-sql-driver/mysql"
 	"pbench/log"
 	"pbench/utils"
+
+	_ "github.com/go-sql-driver/mysql"
 )
 
 var (
@@ -65,7 +66,7 @@ VALUES (?, ?, ?, 0, 0, 0, ?)`
 
 func (m *MySQLRunRecorder) RecordQuery(_ context.Context, s *Stage, result *QueryResult) {
 	recordNewQuery := `INSERT INTO pbench_queries (run_id, stage_id, query_file, query_index, query_id, sequence_no,
-cold_run, succeeded, start_time, end_time, row_count, expected_row_count, duration_ms, info_url) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`
+cold_run, succeeded, start_time, end_time, row_count, expected_row_count, duration_ms, info_url, seed) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`
 	var queryFile string
 	if result.Query.File != nil {
 		queryFile = *result.Query.File
@@ -83,11 +84,13 @@ cold_run, succeeded, start_time, end_time, row_count, expected_row_count, durati
 		result.RowCount, sql.NullInt32{
 			Int32: int32(result.Query.ExpectedRowCount),
 			Valid: result.Query.ExpectedRowCount >= 0,
-		}, result.Duration.Milliseconds(), result.InfoUrl)
+		}, result.Duration.Milliseconds(), result.InfoUrl, result.Seed)
+	log.Info().Str("stage_id", result.StageId).Stringer("start_time", result.StartTime).Stringer("end_time", result.EndTime).
+		Str("info_url", result.InfoUrl).Int64("seed", result.Seed).Msg("recorded query result to MySQL")
 	if err != nil {
 		log.Error().EmbedObject(result).Err(err).Msg("failed to send query summary to MySQL")
 	}
-	updateRunInfo := `UPDATE pbench_runs SET start_time = ?, queries_ran = queries_ran + 1, failed = ?, mismatch = ? WHERE run_id = ?`
+	updateRunInfo := `UPDATE pbench_runs SET start_time = ?, queries_ran = queries_ran + 1, failed = ? , mismatch = ? WHERE run_id = ?`
 	res, err := m.db.Exec(updateRunInfo, s.States.RunStartTime, m.failed, m.mismatch, m.runId)
 	if err != nil {
 		log.Error().Err(err).Str("run_name", s.States.RunName).Int64("run_id", m.runId).

stage/pbench_queries_ddl.sql

@@ -14,6 +14,7 @@ create table if not exists pbench_queries
     expected_row_count int          null,
     duration_ms        int          not null,
     info_url           varchar(255) not null,
+    seed               bigint       not null,
     primary key (run_id, stage_id, query_file, query_index, sequence_no)
 )
     partition by hash (`run_id`) partitions 16;

stage/result.go

@@ -1,13 +1,15 @@
 package stage
 
 import (
-	"github.com/rs/zerolog"
 	"pbench/log"
 	"time"
+
+	"github.com/rs/zerolog"
 )
 
 type QueryResult struct {
 	StageId       string
+	Seed          int64
 	Query         *Query
 	QueryId       string
 	InfoUrl       string