Function CreateQRepWorkflowConfig

flow/e2e/test_utils.go:471–504

(
	t *testing.T,
	flowJobName string,
	sourceTable string,
	dstTable string,
	query string,
	dest string,
	stagingPath string,
	setupDst bool,
	syncedAtCol string,
	isDeletedCol string,
)

}

func CreateQRepWorkflowConfig(
	t *testing.T,
	flowJobName string,
	sourceTable string,
	dstTable string,
	query string,
	dest string,
	stagingPath string,
	setupDst bool,
	syncedAtCol string,
	isDeletedCol string,
) *protos.QRepConfig {
	t.Helper()

	return &protos.QRepConfig{
		FlowJobName:                flowJobName,
		WatermarkTable:             sourceTable,
		DestinationTableIdentifier: dstTable,
		SourceName:                 GeneratePostgresPeer(t).Name,
		DestinationName:            dest,
		Query:                      query,
		WatermarkColumn:            "updated_at",
		StagingPath:                stagingPath,
		WriteMode: &protos.QRepWriteMode{
			WriteType: protos.QRepWriteType_QREP_WRITE_MODE_APPEND,
		},
		NumRowsPerPartition:              1000,
		InitialCopyOnly:                  true,
		SyncedAtColName:                  syncedAtCol,
		SetupWatermarkTableOnDestination: setupDst,
		SoftDeleteColName:                isDeletedCol,
		Version:                          shared.InternalVersion_Latest,
	}
}

func RunQRepFlowWorkflow(t *testing.T, tc client.Client, config *protos.QRepConfig) WorkflowRun {
	t.Helper()