<commit_before>package com.netflix.spinnaker.orca.pipeline.persistence.jedis

import java.util.function.Function
import com.fasterxml.jackson.databind.ObjectMapper
import com.netflix.spectator.api.ExtendedRegistry
import com.netflix.spinnaker.orca.config.OrcaConfiguration
import com.netflix.spinnaker.orca.jackson.OrcaObjectMapper
import com.netflix.spinnaker.orca.pipeline.model.*
import com.netflix.spinnaker.orca.pipeline.persistence.ExecutionNotFoundException
import com.netflix.spinnaker.orca.pipeline.persistence.ExecutionRepository
import groovy.transform.CompileDynamic
import groovy.transform.CompileStatic
import groovy.util.logging.Slf4j
import org.springframework.beans.factory.annotation.Autowired
import org.springframework.beans.factory.annotation.Value
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor
import org.springframework.stereotype.Component
import redis.clients.jedis.Jedis
import redis.clients.jedis.JedisCommands
import redis.clients.util.Pool
import rx.Observable
import rx.Scheduler
import rx.functions.Func1
import rx.schedulers.Schedulers

@Component
@Slf4j
@CompileStatic
class JedisExecutionRepository implements ExecutionRepository {

  private final Pool<Jedis> jedisPool
  private final ObjectMapper mapper = new OrcaObjectMapper()
  private final int chunkSize
  private final Scheduler queryAllScheduler
  private final Scheduler queryByAppScheduler

  @Autowired
  JedisExecutionRepository(
    ExtendedRegistry extendedRegistry,
    Pool<Jedis> jedisPool,
    @Value('${threadPool.executionRepository:150}') int threadPoolSize,
    @Value('${chunkSize.executionRepository:75}') int threadPoolChunkSize
  ) {
    this(
      jedisPool,
      Schedulers.from(newFixedThreadPool(extendedRegistry, 10, "QueryAll")),
      Schedulers.from(newFixedThreadPool(extendedRegistry, threadPoolSize, "QueryByApp")),
      threadPoolChunkSize
    )
  }

  JedisExecutionRepository(
    Pool<Jedis> jedisPool,
    Scheduler queryAllScheduler,
    Scheduler queryByAppScheduler,
    int threadPoolChunkSize
  ) {
    this.jedisPool = jedisPool
    this.queryAllScheduler = queryAllScheduler
    this.queryByAppScheduler = queryByAppScheduler
    this.chunkSize = threadPoolChunkSize
  }

  @Override
  void store(Orchestration orchestration) {
    withJedis { Jedis jedis ->
      storeExecutionInternal(jedis, orchestration)
    }
  }

  @Override
  void store(Pipeline pipeline) {
    withJedis { Jedis jedis ->
      storeExecutionInternal(jedis, pipeline)
      jedis.zadd(executionsByPipelineKey(pipeline.pipelineConfigId), pipeline.buildTime, pipeline.id)
    }
  }

  @Override
  void storeStage(PipelineStage stage) {
    withJedis { Jedis jedis ->
      storeStageInternal(jedis, Pipeline, stage)
    }
  }

  @Override
  void storeStage(Stage stage) {
    if (stage instanceof OrchestrationStage) {
      storeStage((OrchestrationStage) stage)
    } else {
      storeStage((PipelineStage) stage)
    }
  }
<commit_msg>attempt to avoid re-saving stage contexts all the time<commit_after>package com.netflix.spinnaker.orca.pipeline.persistence.jedis

import java.util.function.Function
import com.fasterxml.jackson.databind.ObjectMapper
import com.netflix.spectator.api.ExtendedRegistry
import com.netflix.spinnaker.orca.config.OrcaConfiguration
import com.netflix.spinnaker.orca.jackson.OrcaObjectMapper
import com.netflix.spinnaker.orca.pipeline.model.*
import com.netflix.spinnaker.orca.pipeline.persistence.ExecutionNotFoundException
import com.netflix.spinnaker.orca.pipeline.persistence.ExecutionRepository
import groovy.transform.CompileDynamic
import groovy.transform.CompileStatic
import groovy.util.logging.Slf4j
import org.springframework.beans.factory.annotation.Autowired
import org.springframework.beans.factory.annotation.Value
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor
import org.springframework.stereotype.Component
import redis.clients.jedis.Jedis
import redis.clients.jedis.JedisCommands
import redis.clients.util.Pool
import rx.Observable
import rx.Scheduler
import rx.functions.Func1
import rx.schedulers.Schedulers

@Component
@Slf4j
@CompileStatic
class JedisExecutionRepository implements ExecutionRepository {

  private final Pool<Jedis> jedisPool
  private final ObjectMapper mapper = new OrcaObjectMapper()
  private final int chunkSize
  private final Scheduler queryAllScheduler
  private final Scheduler queryByAppScheduler

  @Autowired
  JedisExecutionRepository(
    ExtendedRegistry extendedRegistry,
    Pool<Jedis> jedisPool,
    @Value('${threadPool.executionRepository:150}') int threadPoolSize,
    @Value('${chunkSize.executionRepository:75}') int threadPoolChunkSize
  ) {
    this(
      jedisPool,
      Schedulers.from(newFixedThreadPool(extendedRegistry, 10, "QueryAll")),
      Schedulers.from(newFixedThreadPool(extendedRegistry, threadPoolSize, "QueryByApp")),
      threadPoolChunkSize
    )
  }

  JedisExecutionRepository(
    Pool<Jedis> jedisPool,
    Scheduler queryAllScheduler,
    Scheduler queryByAppScheduler,
    int threadPoolChunkSize
  ) {
    this.jedisPool = jedisPool
    this.queryAllScheduler = queryAllScheduler
    this.queryByAppScheduler = queryByAppScheduler
    this.chunkSize = threadPoolChunkSize
  }

  @Override
  void store(Orchestration orchestration) {
    withJedis { Jedis jedis ->
      storeExecutionInternal(jedis, orchestration)
    }
  }

  @Override
  void store(Pipeline pipeline) {
    withJedis { Jedis jedis ->
      if (pipeline.id) {
        def json = jedis.hget("pipeline:$pipeline.id", "config")
        def existing = mapper.readValue(json, Pipeline)
        Map<String, Map<String, Object>> contexts = [:]
        pipeline.stages.each { stage ->
          contexts[stage.id] = stage.context
          ((PipelineStage) stage).context = existing.stages.find { it.id == stage.id }?.context
        }
        storeExecutionInternal(jedis, pipeline)
        pipeline.stages.each { stage ->
          ((PipelineStage) stage).context = contexts[stage.id]
        }
      } else {
        storeExecutionInternal(jedis, pipeline)
      }
      jedis.zadd(executionsByPipelineKey(pipeline.pipelineConfigId), pipeline.buildTime, pipeline.id)
    }
  }

  @Override
  void storeStage(PipelineStage stage) {
    withJedis { Jedis jedis ->
      storeStageInternal(jedis, Pipeline, stage)
    }
  }

  @Override
  void storeStage(Stage stage) {
    if (stage instanceof OrchestrationStage) {
      storeStage((OrchestrationStage) stage)
    } else {
      storeStage((PipelineStage) stage)
    }
  }

{'fraction_non_alphanumeric': 0.08305445695162762, 'fraction_numerical': 0.0030422878004259203, 'mean_word_length': 4.5113160100586756, 'pattern_counts': {'":': 0, '<': 11, '<?xml version=': 0, '>': 19, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 2}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1122494', 'n_tokens_mistral': 1986, 'n_tokens_neox': 1972, 'n_words': 354}