<commit_before>/**
 * Copyright (c) 2012-2012 Malhar, Inc.
 * All rights reserved.
 */
package com.malhartech.api;

import com.malhartech.annotation.InputPortFieldAnnotation;
import com.malhartech.annotation.OutputPortFieldAnnotation;
import com.malhartech.api.Context.OperatorContext;
import com.malhartech.api.Context.PortContext;
import com.malhartech.api.Operator.InputPort;
import com.malhartech.api.Operator.OutputPort;
import com.malhartech.engine.Operators;
import com.malhartech.stram.StramUtils;
import com.malhartech.util.AttributeMap;
import com.malhartech.util.AttributeMap.DefaultAttributeMap;
import java.io.*;
import java.lang.reflect.Field;
import java.util.Map.Entry;
import java.util.*;
import javax.validation.*;
import org.apache.commons.lang.builder.ToStringBuilder;
import org.apache.commons.lang.builder.ToStringStyle;
import org.apache.hadoop.conf.Configuration;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * DAG contains the logical declarations of operators and streams.
 * It will be serialized and deployed to the cluster, where it is translated into the physical plan.
 */
public class DAG implements Serializable, DAGConstants
{
  private static final long serialVersionUID = -2099729915606048704L;
  private static final Logger LOG = LoggerFactory.getLogger(DAG.class);
  private final Map<String, StreamDecl> streams = new HashMap<String, StreamDecl>();
  private final Map<String, OperatorWrapper> nodes = new HashMap<String, OperatorWrapper>();
  private final List<OperatorWrapper> rootNodes = new ArrayList<OperatorWrapper>();
  private final AttributeMap<DAGConstants> attributes = new DefaultAttributeMap<DAGConstants>();
  private transient int nodeIndex = 0; // used for cycle validation
  private transient Stack<OperatorWrapper> stack = new Stack<OperatorWrapper>(); // used for cycle validation

  public static class ExternalizableModule implements Externalizable
  {
    private Operator module;

    private void set(Operator module)
    {
      this.module = module;
    }

    @Override
    public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException
    {
      int len = in.readInt();
      byte[] bytes = new byte[len];
      in.read(bytes);
      ByteArrayInputStream bis = new ByteArrayInputStream(bytes);
      set((Operator)new DefaultOperatorSerDe().read(bis));
      bis.close();
    }

    @Override
    public void writeExternal(ObjectOutput out) throws IOException
    {
      ByteArrayOutputStream bos = new ByteArrayOutputStream();
      new DefaultOperatorSerDe().write(module, bos);
      bos.close();
      byte[] bytes = bos.toByteArray();
      out.writeInt(bytes.length);
      out.write(bytes);
    }
  }

  public DAG()
  {
  }

  @SuppressWarnings("unchecked")
  public DAG(Configuration conf)
  {
    for (@SuppressWarnings("rawtypes") DAGConstants.AttributeKey key : DAGConstants.ATTRIBUTE_KEYS) {
      String stringValue = conf.get(key.name());
      if (stringValue != null) {
        if (key.attributeType == Integer.class) {
          this.attributes.attr((DAGConstants.AttributeKey<Integer>)key).set(conf.getInt(key.name(), 0));
        } else if (key.attributeType == Long.class) {
          this.attributes.attr((DAGConstants.AttributeKey<Long>)key).set(conf.getLong(key.name(), 0));
        } else if (key.attributeType == String.class) {
          this.attributes.attr((DAGConstants.AttributeKey<String>)key).set(stringValue);
        } else if (key.attributeType == Boolean.class) {
          this.attributes.attr((DAGConstants.AttributeKey<Boolean>)key).set(conf.getBoolean(key.name(), false));
        } else {
          String msg = String.format("Unsupported attribute type: %s (%s)", key.attributeType, key.name());
          throw new UnsupportedOperationException(msg);
        }
      }
    }
  }

  public final class InputPortMeta implements Serializable
  {
    private static final long serialVersionUID = 1L;
    private OperatorWrapper operatorWrapper;
    private String fieldName;
    private InputPortFieldAnnotation portAnnotation;
    private final AttributeMap<PortContext> attributes = new DefaultAttributeMap<PortContext>();

    public OperatorWrapper getOperatorWrapper()
    {
      return operatorWrapper;
    }

    public String getPortName()
    {
      return portAnnotation == null || portAnnotation.name() == null ? fieldName : portAnnotation.name();
    }

    public InputPort<?> getPortObject() {
      for (Entry<InputPort<?>, InputPortMeta> e : operatorWrapper.getPortMapping().inPortMap.entrySet()) {
        if (e.getValue() == this) {
          return e.getKey();
        }
      }
      throw new AssertionError();
    }

    public AttributeMap<PortContext> getAttributes() {
      return attributes;
    }

    @Override
    public String toString()
    {<commit_msg>Add a message to the assertion error.<commit_after>/**
 * Copyright (c) 2012-2012 Malhar, Inc.
 * All rights reserved.
 */
package com.malhartech.api;

import com.malhartech.annotation.InputPortFieldAnnotation;
import com.malhartech.annotation.OutputPortFieldAnnotation;
import com.malhartech.api.Context.OperatorContext;
import com.malhartech.api.Context.PortContext;
import com.malhartech.api.Operator.InputPort;
import com.malhartech.api.Operator.OutputPort;
import com.malhartech.engine.Operators;
import com.malhartech.stram.StramUtils;
import com.malhartech.util.AttributeMap;
import com.malhartech.util.AttributeMap.DefaultAttributeMap;
import java.io.*;
import java.lang.reflect.Field;
import java.util.Map.Entry;
import java.util.*;
import javax.validation.*;
import org.apache.commons.lang.builder.ToStringBuilder;
import org.apache.commons.lang.builder.ToStringStyle;
import org.apache.hadoop.conf.Configuration;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * DAG contains the logical declarations of operators and streams.
 * It will be serialized and deployed to the cluster, where it is translated into the physical plan.
 */
public class DAG implements Serializable, DAGConstants
{
  private static final long serialVersionUID = -2099729915606048704L;
  private static final Logger LOG = LoggerFactory.getLogger(DAG.class);
  private final Map<String, StreamDecl> streams = new HashMap<String, StreamDecl>();
  private final Map<String, OperatorWrapper> nodes = new HashMap<String, OperatorWrapper>();
  private final List<OperatorWrapper> rootNodes = new ArrayList<OperatorWrapper>();
  private final AttributeMap<DAGConstants> attributes = new DefaultAttributeMap<DAGConstants>();
  private transient int nodeIndex = 0; // used for cycle validation
  private transient Stack<OperatorWrapper> stack = new Stack<OperatorWrapper>(); // used for cycle validation

  public static class ExternalizableModule implements Externalizable
  {
    private Operator module;

    private void set(Operator module)
    {
      this.module = module;
    }

    @Override
    public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException
    {
      int len = in.readInt();
      byte[] bytes = new byte[len];
      in.read(bytes);
      ByteArrayInputStream bis = new ByteArrayInputStream(bytes);
      set((Operator)new DefaultOperatorSerDe().read(bis));
      bis.close();
    }

    @Override
    public void writeExternal(ObjectOutput out) throws IOException
    {
      ByteArrayOutputStream bos = new ByteArrayOutputStream();
      new DefaultOperatorSerDe().write(module, bos);
      bos.close();
      byte[] bytes = bos.toByteArray();
      out.writeInt(bytes.length);
      out.write(bytes);
    }
  }

  public DAG()
  {
  }

  @SuppressWarnings("unchecked")
  public DAG(Configuration conf)
  {
    for (@SuppressWarnings("rawtypes") DAGConstants.AttributeKey key : DAGConstants.ATTRIBUTE_KEYS) {
      String stringValue = conf.get(key.name());
      if (stringValue != null) {
        if (key.attributeType == Integer.class) {
          this.attributes.attr((DAGConstants.AttributeKey<Integer>)key).set(conf.getInt(key.name(), 0));
        } else if (key.attributeType == Long.class) {
          this.attributes.attr((DAGConstants.AttributeKey<Long>)key).set(conf.getLong(key.name(), 0));
        } else if (key.attributeType == String.class) {
          this.attributes.attr((DAGConstants.AttributeKey<String>)key).set(stringValue);
        } else if (key.attributeType == Boolean.class) {
          this.attributes.attr((DAGConstants.AttributeKey<Boolean>)key).set(conf.getBoolean(key.name(), false));
        } else {
          String msg = String.format("Unsupported attribute type: %s (%s)", key.attributeType, key.name());
          throw new UnsupportedOperationException(msg);
        }
      }
    }
  }

  public final class InputPortMeta implements Serializable
  {
    private static final long serialVersionUID = 1L;
    private OperatorWrapper operatorWrapper;
    private String fieldName;
    private InputPortFieldAnnotation portAnnotation;
    private final AttributeMap<PortContext> attributes = new DefaultAttributeMap<PortContext>();

    public OperatorWrapper getOperatorWrapper()
    {
      return operatorWrapper;
    }

    public String getPortName()
    {
      return portAnnotation == null || portAnnotation.name() == null ? fieldName : portAnnotation.name();
    }

    public InputPort<?> getPortObject() {
      for (Entry<InputPort<?>, InputPortMeta> e : operatorWrapper.getPortMapping().inPortMap.entrySet()) {
        if (e.getValue() == this) {
          return e.getKey();
        }
      }
      throw new AssertionError("Cannot find the port object for " + this);
    }

    public AttributeMap<PortContext> getAttributes() {
      return attributes;
    }

    @Override
    public String toString()
    {
{'fraction_non_alphanumeric': 0.10717585463504774, 'fraction_numerical': 0.006775485063135201, 'mean_word_length': 3.875875875875876, 'pattern_counts': {'":': 0, '<': 43, '<?xml version=': 0, '>': 43, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, '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': '927551', 'n_tokens_mistral': 2682, 'n_tokens_neox': 2628, 'n_words': 650}