<commit_before>package com.continuuity.internal.app.runtime.batch;

import com.continuuity.api.data.batch.BatchReadable;
import com.continuuity.api.data.batch.BatchWritable;
import com.continuuity.api.data.batch.Split;
import com.continuuity.api.mapreduce.MapReduceContext;
import com.continuuity.api.mapreduce.MapReduceSpecification;
import com.continuuity.api.metrics.Metrics;
import com.continuuity.app.metrics.MapReduceMetrics;
import com.continuuity.app.program.Program;
import com.continuuity.app.runtime.Arguments;
import com.continuuity.common.logging.LoggingContext;
import com.continuuity.common.metrics.MetricsCollectionService;
import com.continuuity.common.metrics.MetricsCollector;
import com.continuuity.common.metrics.MetricsScope;
import com.continuuity.data2.transaction.TransactionAware;
import com.continuuity.internal.app.runtime.AbstractContext;
import com.continuuity.internal.app.runtime.ProgramServiceDiscovery;
import com.continuuity.logging.context.MapReduceLoggingContext;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Maps;
import org.apache.hadoop.mapreduce.Job;
import org.apache.twill.api.RunId;
import org.apache.twill.discovery.ServiceDiscovered;

import java.io.Closeable;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import javax.annotation.Nullable;

/**
 * Mapreduce job runtime context
 */
public class BasicMapReduceContext extends AbstractContext implements MapReduceContext {

  private final String accountId;
  private final MapReduceSpecification spec;
  private final MapReduceLoggingContext loggingContext;
  private final Map<MetricsScope, MetricsCollector> systemMapperMetrics;
  private final Map<MetricsScope, MetricsCollector> systemReducerMetrics;
  private final Map<MetricsScope, MetricsCollector> systemMetrics;
  private final Arguments runtimeArguments;
  private final long logicalStartTime;
  private final String workflowBatch;
  private final Metrics mapredMetrics;
  private final MetricsCollectionService metricsCollectionService;
  private final ProgramServiceDiscovery serviceDiscovery;

  /**
   * Input dataset for the MapReduce job. May be set be either inputDataset or inputDatasetName, but not both.
   */
  private BatchReadable inputDataset;
  private String inputDatasetName;
  private List<Split> inputDataSelection;

  /**
   * Output dataset for the MapReduce job. May be set by either outputDataset or outputDatasetName, but not both.
   */
  private BatchWritable outputDataset;
  private String outputDatasetName;
  private Job job;

  // todo: having it here seems like a hack will be fixed with further post-integration refactoring
  private final Iterable<TransactionAware> txAwares;

  public BasicMapReduceContext(Program program,
                               MapReduceMetrics.TaskType type,
                               RunId runId,
                               Arguments runtimeArguments,
                               Map<String, Closeable> datasets,
                               MapReduceSpecification spec,
                               Iterable<TransactionAware> txAwares,
                               long logicalStartTime,
                               String workflowBatch,
                               ProgramServiceDiscovery serviceDiscovery) {
    this(program, type, runId, runtimeArguments, datasets,
         spec, txAwares, logicalStartTime, workflowBatch, serviceDiscovery, null);
  }


  public BasicMapReduceContext(Program program,
                               MapReduceMetrics.TaskType type,
                               RunId runId,
                               Arguments runtimeArguments,
                               Map<String, Closeable> datasets,
                               MapReduceSpecification spec,
                               Iterable<TransactionAware> txAwares,
                               long logicalStartTime,
                               String workflowBatch,
                               ProgramServiceDiscovery serviceDiscovery,
                               MetricsCollectionService metricsCollectionService) {
    super(program, runId, datasets);
    this.accountId = program.getAccountId();
    this.runtimeArguments = runtimeArguments;
    this.logicalStartTime = logicalStartTime;
    this.workflowBatch = workflowBatch;
    this.serviceDiscovery = serviceDiscovery;
    this.metricsCollectionService = metricsCollectionService;

    if (metricsCollectionService != null) {
      this.systemMapperMetrics = Maps.newHashMap();
      this.systemReducerMetrics = Maps.newHashMap();
      this.systemMetrics = Maps.newHashMap();
      for (MetricsScope scope : MetricsScope.values()) {
        this.systemMapperMetrics.put(scope, getMetricsCollector(scope, metricsCollectionService,
                                                                getMetricContext(MapReduceMetrics.TaskType.Mapper)));
        this.systemReducerMetrics.put(scope, getMetricsCollector(scope, metricsCollectionService,
                                                                 getMetricContext(MapReduceMetrics.TaskType.Reducer)));
        this.systemMetrics.put(scope, getMetricsCollector(scope, metricsCollectionService,
                                                                 getMetricContext()));
      }
      // for user metrics.  type can be null if its not in a map or reduce task, but in the yarn container that
      // launches the mapred job.
      this.mapredMetrics = (type == null) ?
        null : new MapReduceMetrics(metricsCollectionService, getApplicationId(), getProgramName(), type);
    } else {
      this.systemMapperMetrics = null;
      this.systemReducerMetrics = null;
      this.systemMetrics = null;
      this.mapredMetrics = null;
    }
    this.loggingContext = new MapReduceLoggingContext(getAccountId(), getApplicationId(), getProgramName());
    this.spec = spec;
    this.txAwares = txAwares;
  }

  @Override
  public String toString() {
    return String.format("job=%s,=%s",
                         spec.getName(), super.toString());
  }

  @Override
  public MapReduceSpecification getSpecification() {
    return spec;
  }

  @Override
  public long getLogicalStartTime() {
    return logicalStartTime;
  }

  /**
   * Returns the name of the Batch job when running inside workflow. Otherwise, return null.
   */
  public String getWorkflowBatch() {
    return workflowBatch;
  }

  public void setJob(Job job) {
    this.job = job;
  }

  @SuppressWarnings("unchecked")
  @Override
  public <T> T getHadoopJob() {
    return (T) job;
  }

  @Override
  public void setInput(BatchReadable dataset, List<Split> splits) {
    this.inputDataset = dataset;
    this.inputDatasetName = null;
    this.inputDataSelection = splits;
  }

  @Override
  public void setInput(String datasetName, List<Split> splits) {
    this.inputDataset = null;
    this.inputDatasetName = datasetName;
    this.inputDataSelection = splits;
  }

  @Override<commit_msg>Added @Deprecated to BasicMapReduceContext<commit_after>package com.continuuity.internal.app.runtime.batch;

import com.continuuity.api.data.batch.BatchReadable;
import com.continuuity.api.data.batch.BatchWritable;
import com.continuuity.api.data.batch.Split;
import com.continuuity.api.mapreduce.MapReduceContext;
import com.continuuity.api.mapreduce.MapReduceSpecification;
import com.continuuity.api.metrics.Metrics;
import com.continuuity.app.metrics.MapReduceMetrics;
import com.continuuity.app.program.Program;
import com.continuuity.app.runtime.Arguments;
import com.continuuity.common.logging.LoggingContext;
import com.continuuity.common.metrics.MetricsCollectionService;
import com.continuuity.common.metrics.MetricsCollector;
import com.continuuity.common.metrics.MetricsScope;
import com.continuuity.data2.transaction.TransactionAware;
import com.continuuity.internal.app.runtime.AbstractContext;
import com.continuuity.internal.app.runtime.ProgramServiceDiscovery;
import com.continuuity.logging.context.MapReduceLoggingContext;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Maps;
import org.apache.hadoop.mapreduce.Job;
import org.apache.twill.api.RunId;
import org.apache.twill.discovery.ServiceDiscovered;

import java.io.Closeable;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import javax.annotation.Nullable;

/**
 * Mapreduce job runtime context
 */
public class BasicMapReduceContext extends AbstractContext implements MapReduceContext {

  private final String accountId;
  private final MapReduceSpecification spec;
  private final MapReduceLoggingContext loggingContext;
  private final Map<MetricsScope, MetricsCollector> systemMapperMetrics;
  private final Map<MetricsScope, MetricsCollector> systemReducerMetrics;
  private final Map<MetricsScope, MetricsCollector> systemMetrics;
  private final Arguments runtimeArguments;
  private final long logicalStartTime;
  private final String workflowBatch;
  private final Metrics mapredMetrics;
  private final MetricsCollectionService metricsCollectionService;
  private final ProgramServiceDiscovery serviceDiscovery;

  /**
   * Input dataset for the MapReduce job. Either inputDataset or inputDatasetName is set.
   */
  @Deprecated
  private BatchReadable inputDataset;
  private String inputDatasetName;
  private List<Split> inputDataSelection;

  /**
   * Output dataset for the MapReduce job. Either outputDataset or outputDatasetName is set.
   */
  @Deprecated
  private BatchWritable outputDataset;
  private String outputDatasetName;
  private Job job;

  // todo: having it here seems like a hack will be fixed with further post-integration refactoring
  private final Iterable<TransactionAware> txAwares;

  public BasicMapReduceContext(Program program,
                               MapReduceMetrics.TaskType type,
                               RunId runId,
                               Arguments runtimeArguments,
                               Map<String, Closeable> datasets,
                               MapReduceSpecification spec,
                               Iterable<TransactionAware> txAwares,
                               long logicalStartTime,
                               String workflowBatch,
                               ProgramServiceDiscovery serviceDiscovery) {
    this(program, type, runId, runtimeArguments, datasets,
         spec, txAwares, logicalStartTime, workflowBatch, serviceDiscovery, null);
  }


  public BasicMapReduceContext(Program program,
                               MapReduceMetrics.TaskType type,
                               RunId runId,
                               Arguments runtimeArguments,
                               Map<String, Closeable> datasets,
                               MapReduceSpecification spec,
                               Iterable<TransactionAware> txAwares,
                               long logicalStartTime,
                               String workflowBatch,
                               ProgramServiceDiscovery serviceDiscovery,
                               MetricsCollectionService metricsCollectionService) {
    super(program, runId, datasets);
    this.accountId = program.getAccountId();
    this.runtimeArguments = runtimeArguments;
    this.logicalStartTime = logicalStartTime;
    this.workflowBatch = workflowBatch;
    this.serviceDiscovery = serviceDiscovery;
    this.metricsCollectionService = metricsCollectionService;

    if (metricsCollectionService != null) {
      this.systemMapperMetrics = Maps.newHashMap();
      this.systemReducerMetrics = Maps.newHashMap();
      this.systemMetrics = Maps.newHashMap();
      for (MetricsScope scope : MetricsScope.values()) {
        this.systemMapperMetrics.put(scope, getMetricsCollector(scope, metricsCollectionService,
                                                                getMetricContext(MapReduceMetrics.TaskType.Mapper)));
        this.systemReducerMetrics.put(scope, getMetricsCollector(scope, metricsCollectionService,
                                                                 getMetricContext(MapReduceMetrics.TaskType.Reducer)));
        this.systemMetrics.put(scope, getMetricsCollector(scope, metricsCollectionService,
                                                                 getMetricContext()));
      }
      // for user metrics.  type can be null if its not in a map or reduce task, but in the yarn container that
      // launches the mapred job.
      this.mapredMetrics = (type == null) ?
        null : new MapReduceMetrics(metricsCollectionService, getApplicationId(), getProgramName(), type);
    } else {
      this.systemMapperMetrics = null;
      this.systemReducerMetrics = null;
      this.systemMetrics = null;
      this.mapredMetrics = null;
    }
    this.loggingContext = new MapReduceLoggingContext(getAccountId(), getApplicationId(), getProgramName());
    this.spec = spec;
    this.txAwares = txAwares;
  }

  @Override
  public String toString() {
    return String.format("job=%s,=%s",
                         spec.getName(), super.toString());
  }

  @Override
  public MapReduceSpecification getSpecification() {
    return spec;
  }

  @Override
  public long getLogicalStartTime() {
    return logicalStartTime;
  }

  /**
   * Returns the name of the Batch job when running inside workflow. Otherwise, return null.
   */
  public String getWorkflowBatch() {
    return workflowBatch;
  }

  public void setJob(Job job) {
    this.job = job;
  }

  @SuppressWarnings("unchecked")
  @Override
  public <T> T getHadoopJob() {
    return (T) job;
  }

  @Deprecated
  @Override
  public void setInput(BatchReadable dataset, List<Split> splits) {
    this.inputDataset = dataset;
    this.inputDatasetName = null;
    this.inputDataSelection = splits;
  }

  @Override
  public void setInput(String datasetName, List<Split> splits) {
    this.inputDataset = null;
    this.inputDatasetName = datasetName;
    this.inputDataSelection = splits;
  }

  @Deprecated
  @Override
{'fraction_non_alphanumeric': 0.07347606515399388, 'fraction_numerical': 0.00014225762856533183, 'mean_word_length': 2.8289760348583877, 'pattern_counts': {'":': 0, '<': 27, '<?xml version=': 0, '>': 27, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 12, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2410034', 'n_tokens_mistral': 3602, 'n_tokens_neox': 3530, 'n_words': 836}