<commit_before>package water.fvec;

import water.*;
import water.H2O.H2OCountedCompleter;
import water.exec.Flow;
import water.util.Log;

import java.io.IOException;
import java.io.InputStream;
import java.util.Arrays;
import java.util.HashMap;
import java.util.IllegalFormatException;

/**
 * A collection of named Vecs.  Essentially an R-like data-frame.  Multiple
 * Frames can reference the same Vecs.  A Frame is a lightweight object, it is
 * meant to be cheaply created and discarded for data munging purposes.
 * E.g. to exclude a Vec from a computation on a Frame, create a new Frame that
 * references all the Vecs but this one.
 */
public class Frame extends Lockable<Frame> {
  public String[] _names;
  Key[] _keys;          // Keys for the vectors
  private transient Vec[] _vecs;// The Vectors (transient to avoid network traffic)
  private transient Vec _col0;  // First readable vec; fast access to the VectorGroup's Chunk layout
  private final UniqueId uniqueId;

  public Frame(Key k){
    super(k);
    uniqueId = new UniqueFrameId(k, this);
  }
  public Frame( Frame fr ) { this(fr._key,fr._names.clone(), fr.vecs().clone()); _col0 = null; }
  public Frame( Vec... vecs ){ this(null,vecs);}
  public Frame( String[] names, Vec[] vecs ) { this(null,names,vecs); }

  public Frame( Key key, String[] names, Vec[] vecs ) {
    super(key);
    this.uniqueId = new UniqueFrameId(_key, this);
    if( names==null ) {
      names = new String[vecs.length];
      for( int i=0; i<vecs.length; i++ ) names[i] = "C"+(i+1);
    }
    assert names.length == vecs.length : "Number of columns does not match to number of cols' names.";
    _names=names;
    _vecs=vecs;
    _keys = new Key[vecs.length];
    for( int i=0; i<vecs.length; i++ ) {
      Key k = _keys[i] = vecs[i]._key;
      if( DKV.get(k)==null )    // If not already in KV, put it there
        DKV.put(k,vecs[i]);
    }
    assert checkCompatible();
  }

  public UniqueId getUniqueId() {
    return this.uniqueId;
  }

  /** 64-bit checksum of the checksums of the vecs.  SHA-265 checksums of the chunks are XORed
   * together.  Since parse always parses the same pieces of files into the same offsets
   * in some chunk this checksum will be consistent across reparses.
   */
  public long checksum() {
    Vec [] vecs = vecs();
    long _checksum = 0;
    for(int i = 0; i < _names.length; ++i) {
      long vec_checksum = vecs[i].checksum();
      _checksum ^= vec_checksum;
      _checksum ^= (2147483647 * i);
    }
    return _checksum;
  }

  public Vec vec(String name){
    Vec [] vecs = vecs();
    for(int i = 0; i < _names.length; ++i)
      if(_names[i].equals(name))return vecs[i];
    return null;
  }
  /** Returns the vector by given index.
   * <p>The call is direct equivalent to call <code>vecs()[i]</code> and
   * it does not do any array bounds checking.</p>
   * @param idx idx of column
   * @return this frame idx-th vector, never returns <code>null</code>
   */
  public Vec vec(int idx) {
    Vec[] vecs = vecs();
    return vecs[idx];
  }
  /** Returns a subframe of this frame containing only vectors with desired names.
   *
   * @param names list of vector names
   * @return a new frame which collects vectors from this frame with desired names.
   * @throws IllegalArgumentException if there is no vector with desired name in this frame.
   */
  public Frame subframe(String[] names) { return subframe(names, false, 0)[0]; }
  /** Returns a new frame composed of vectors of this frame selected by given names.
   * The method replaces missing vectors by a constant column filled by given value.
   * @param names names of vector to compose a subframe
   * @param c value to fill missing columns.
   * @return two frames, the first contains subframe, the second contains newly created constant vectors or null
   */
  public Frame[] subframe(String[] names, double c) { return subframe(names, true, c); }
  /** Create a subframe from this frame based on desired names.
   * Throws an exception if desired column is not in this frame and <code>replaceBy</code> is <code>false</code>.
   * Else replace a missing column by a constant column with given value.
   *
   * @param names list of column names to extract
   * @param replaceBy should be missing column replaced by a constant column
   * @param c value for constant column
   * @return array of 2 frames, the first is containing a desired subframe, the second one contains newly created columns or null
   * @throws IllegalArgumentException if <code>replaceBy</code> is false and there is a missing column in this frame
   */
  private Frame[] subframe(String[] names, boolean replaceBy, double c){
    Vec [] vecs     = new Vec[names.length];
    Vec [] cvecs    = replaceBy ? new Vec   [names.length] : null;
    String[] cnames = replaceBy ? new String[names.length] : null;
    int ccv = 0; // counter of constant columns
    vecs();                     // Preload the vecs
    HashMap<String, Integer> map = new HashMap<String, Integer>((int) ((names.length/0.75f)+1)); // avoid rehashing by set up initial capacity
    for(int i = 0; i < _names.length; ++i) map.put(_names[i], i);
    for(int i = 0; i < names.length; ++i)
      if(map.containsKey(names[i])) vecs[i] = _vecs[map.get(names[i])];
      else if (replaceBy) {
        Log.warn("Column " + names[i] + " is missing, filling it in with " + c);
        cnames[ccv] = names[i];
        vecs[i] = cvecs[ccv++] = anyVec().makeCon(c);
      }
    return new Frame[] { new Frame(names,vecs), ccv>0 ?  new Frame(Arrays.copyOf(cnames, ccv), Arrays.copyOf(cvecs,ccv)) : null };
  }

  // Return (and cache) vectors
  public final Vec[] vecs() {
    Vec[] tvecs = _vecs; // read the content
    return tvecs == null ? (_vecs=vecs_impl()) : tvecs;
  }
  // Compute vectors for caching
  private Vec[] vecs_impl() {
    // Load all Vec headers; load them all in parallel by spawning F/J tasks.
    final Vec [] vecs = new Vec[_keys.length];
    Futures fs = new Futures();
    for( int i=0; i<_keys.length; i++ ) {
      final int ii = i;
      final Key k = _keys[i];
      H2OCountedCompleter t = new H2OCountedCompleter() {
          // We need higher priority here as there is a danger of deadlock in
          // case of many calls from MRTask2 at once (e.g. frame with many
          // vectors invokes rollup tasks for all vectors in parallel).  Should
          // probably be done in CPS style in the future
          @Override public byte priority(){return H2O.MIN_HI_PRIORITY;}
          @Override public void compute2() {
            Value v = DKV.get(k);
            if( v==null ) Log.err("Missing vector #" + ii + " (" + _names[ii] + ") during Frame fetch: "+k);
            vecs[ii] = v.get();
            tryComplete();
          }
        };
      H2O.submitTask(t);
      fs.add(t);
    }
    fs.blockForPending();
    return vecs;
  }
  // Force a cache-flush & reload, assuming vec mappings were altered remotely
  public final Vec[] reloadVecs() { _vecs=null; return vecs(); }

  /** Finds the first column with a matching name.  */
  public int find( String name ) {
    if (_names!=null)
      for( int i=0; i<_names.length; i++ )
        if( name.equals(_names[i]) )
          return i;
    return -1;
  }

  public int find( Vec vec ) {
    for( int i=0; i<_vecs.length; i++ )
      if( vec.equals(_vecs[i]) )
        return i;
    return -1;
  }

  // Return Frame 'f' if 'f' is compatible with 'this'.
  // Return a new Frame compatible with 'this' and a copy of 'f's data otherwise.
  public Frame makeCompatible( Frame f) {
    // Small data frames are always "compatible"
    if( anyVec()==null ||       // No dest columns
        numRows() <= 1e4 )      // Or it is small
      return f;                 // Then must be compatible
    // Same VectorGroup is also compatible
    if( f.anyVec() == null ||
        f.anyVec().group().equals(anyVec().group()) )
      return f;
    // Ok, here make some new Vecs with compatible layout
    Key k = Key.make();
    H2O.submitTask(new RebalanceDataSet(this, f, k)).join();
    Frame f2 = DKV.get(k).get();
    DKV.remove(k);
    return f2;
  }

 /** Appends a named column, keeping the last Vec as the response */
  public Frame add( String name, Vec vec ) {
    if( find(name) != -1 ) throw new IllegalArgumentException("Duplicate name '"+name+"' in Frame");
    if( _vecs.length != 0 ) {
      if( !anyVec().group().equals(vec.group()) && !Arrays.equals(anyVec()._espc,vec._espc) )
        throw new IllegalArgumentException("Vector groups differs - adding vec '"+name+"' into the frame " + Arrays.toString(_names));
      if( numRows() != vec.length() )
        throw new IllegalArgumentException("Vector lengths differ - adding vec '"+name+"' into the frame " + Arrays.toString(_names));
    }
    final int len = _names != null ? _names.length : 0;
    _names = _names != null ? Arrays.copyOf(_names,len+1) : new String[len+1];
    _vecs  = _names != null ? Arrays.copyOf(_vecs ,len+1) : new Vec   [len+1];
    _keys  = _names != null ? Arrays.copyOf(_keys ,len+1) : new Key   [len+1];
    _names[len] = name;<commit_msg>Bugfix in makeCompatible call on a frame:   1) do not skip the call if small (there was a bail if <1e4)   2) also check _espc when checking for compatibility (only tested vector group which is not enough)<commit_after>package water.fvec;

import water.*;
import water.H2O.H2OCountedCompleter;
import water.exec.Flow;
import water.util.Log;

import java.io.IOException;
import java.io.InputStream;
import java.util.Arrays;
import java.util.HashMap;
import java.util.IllegalFormatException;

/**
 * A collection of named Vecs.  Essentially an R-like data-frame.  Multiple
 * Frames can reference the same Vecs.  A Frame is a lightweight object, it is
 * meant to be cheaply created and discarded for data munging purposes.
 * E.g. to exclude a Vec from a computation on a Frame, create a new Frame that
 * references all the Vecs but this one.
 */
public class Frame extends Lockable<Frame> {
  public String[] _names;
  Key[] _keys;          // Keys for the vectors
  private transient Vec[] _vecs;// The Vectors (transient to avoid network traffic)
  private transient Vec _col0;  // First readable vec; fast access to the VectorGroup's Chunk layout
  private final UniqueId uniqueId;

  public Frame(Key k){
    super(k);
    uniqueId = new UniqueFrameId(k, this);
  }
  public Frame( Frame fr ) { this(fr._key,fr._names.clone(), fr.vecs().clone()); _col0 = null; }
  public Frame( Vec... vecs ){ this(null,vecs);}
  public Frame( String[] names, Vec[] vecs ) { this(null,names,vecs); }

  public Frame( Key key, String[] names, Vec[] vecs ) {
    super(key);
    this.uniqueId = new UniqueFrameId(_key, this);
    if( names==null ) {
      names = new String[vecs.length];
      for( int i=0; i<vecs.length; i++ ) names[i] = "C"+(i+1);
    }
    assert names.length == vecs.length : "Number of columns does not match to number of cols' names.";
    _names=names;
    _vecs=vecs;
    _keys = new Key[vecs.length];
    for( int i=0; i<vecs.length; i++ ) {
      Key k = _keys[i] = vecs[i]._key;
      if( DKV.get(k)==null )    // If not already in KV, put it there
        DKV.put(k,vecs[i]);
    }
    assert checkCompatible();
  }

  public UniqueId getUniqueId() {
    return this.uniqueId;
  }

  /** 64-bit checksum of the checksums of the vecs.  SHA-265 checksums of the chunks are XORed
   * together.  Since parse always parses the same pieces of files into the same offsets
   * in some chunk this checksum will be consistent across reparses.
   */
  public long checksum() {
    Vec [] vecs = vecs();
    long _checksum = 0;
    for(int i = 0; i < _names.length; ++i) {
      long vec_checksum = vecs[i].checksum();
      _checksum ^= vec_checksum;
      _checksum ^= (2147483647 * i);
    }
    return _checksum;
  }

  public Vec vec(String name){
    Vec [] vecs = vecs();
    for(int i = 0; i < _names.length; ++i)
      if(_names[i].equals(name))return vecs[i];
    return null;
  }
  /** Returns the vector by given index.
   * <p>The call is direct equivalent to call <code>vecs()[i]</code> and
   * it does not do any array bounds checking.</p>
   * @param idx idx of column
   * @return this frame idx-th vector, never returns <code>null</code>
   */
  public Vec vec(int idx) {
    Vec[] vecs = vecs();
    return vecs[idx];
  }
  /** Returns a subframe of this frame containing only vectors with desired names.
   *
   * @param names list of vector names
   * @return a new frame which collects vectors from this frame with desired names.
   * @throws IllegalArgumentException if there is no vector with desired name in this frame.
   */
  public Frame subframe(String[] names) { return subframe(names, false, 0)[0]; }
  /** Returns a new frame composed of vectors of this frame selected by given names.
   * The method replaces missing vectors by a constant column filled by given value.
   * @param names names of vector to compose a subframe
   * @param c value to fill missing columns.
   * @return two frames, the first contains subframe, the second contains newly created constant vectors or null
   */
  public Frame[] subframe(String[] names, double c) { return subframe(names, true, c); }
  /** Create a subframe from this frame based on desired names.
   * Throws an exception if desired column is not in this frame and <code>replaceBy</code> is <code>false</code>.
   * Else replace a missing column by a constant column with given value.
   *
   * @param names list of column names to extract
   * @param replaceBy should be missing column replaced by a constant column
   * @param c value for constant column
   * @return array of 2 frames, the first is containing a desired subframe, the second one contains newly created columns or null
   * @throws IllegalArgumentException if <code>replaceBy</code> is false and there is a missing column in this frame
   */
  private Frame[] subframe(String[] names, boolean replaceBy, double c){
    Vec [] vecs     = new Vec[names.length];
    Vec [] cvecs    = replaceBy ? new Vec   [names.length] : null;
    String[] cnames = replaceBy ? new String[names.length] : null;
    int ccv = 0; // counter of constant columns
    vecs();                     // Preload the vecs
    HashMap<String, Integer> map = new HashMap<String, Integer>((int) ((names.length/0.75f)+1)); // avoid rehashing by set up initial capacity
    for(int i = 0; i < _names.length; ++i) map.put(_names[i], i);
    for(int i = 0; i < names.length; ++i)
      if(map.containsKey(names[i])) vecs[i] = _vecs[map.get(names[i])];
      else if (replaceBy) {
        Log.warn("Column " + names[i] + " is missing, filling it in with " + c);
        cnames[ccv] = names[i];
        vecs[i] = cvecs[ccv++] = anyVec().makeCon(c);
      }
    return new Frame[] { new Frame(names,vecs), ccv>0 ?  new Frame(Arrays.copyOf(cnames, ccv), Arrays.copyOf(cvecs,ccv)) : null };
  }

  // Return (and cache) vectors
  public final Vec[] vecs() {
    Vec[] tvecs = _vecs; // read the content
    return tvecs == null ? (_vecs=vecs_impl()) : tvecs;
  }
  // Compute vectors for caching
  private Vec[] vecs_impl() {
    // Load all Vec headers; load them all in parallel by spawning F/J tasks.
    final Vec [] vecs = new Vec[_keys.length];
    Futures fs = new Futures();
    for( int i=0; i<_keys.length; i++ ) {
      final int ii = i;
      final Key k = _keys[i];
      H2OCountedCompleter t = new H2OCountedCompleter() {
          // We need higher priority here as there is a danger of deadlock in
          // case of many calls from MRTask2 at once (e.g. frame with many
          // vectors invokes rollup tasks for all vectors in parallel).  Should
          // probably be done in CPS style in the future
          @Override public byte priority(){return H2O.MIN_HI_PRIORITY;}
          @Override public void compute2() {
            Value v = DKV.get(k);
            if( v==null ) Log.err("Missing vector #" + ii + " (" + _names[ii] + ") during Frame fetch: "+k);
            vecs[ii] = v.get();
            tryComplete();
          }
        };
      H2O.submitTask(t);
      fs.add(t);
    }
    fs.blockForPending();
    return vecs;
  }
  // Force a cache-flush & reload, assuming vec mappings were altered remotely
  public final Vec[] reloadVecs() { _vecs=null; return vecs(); }

  /** Finds the first column with a matching name.  */
  public int find( String name ) {
    if (_names!=null)
      for( int i=0; i<_names.length; i++ )
        if( name.equals(_names[i]) )
          return i;
    return -1;
  }

  public int find( Vec vec ) {
    for( int i=0; i<_vecs.length; i++ )
      if( vec.equals(_vecs[i]) )
        return i;
    return -1;
  }

  // Return Frame 'f' if 'f' is compatible with 'this'.
  // Return a new Frame compatible with 'this' and a copy of 'f's data otherwise.
  public Frame makeCompatible( Frame f) {
    // Small data frames are always "compatible"
    if( anyVec()==null)      // Or it is small
      return f;                 // Then must be compatible
    // Same VectorGroup is also compatible
    if( f.anyVec() == null ||
        f.anyVec().group().equals(anyVec().group()) && Arrays.equals(f.anyVec()._espc,anyVec()._espc))
      return f;
    // Ok, here make some new Vecs with compatible layout
    Key k = Key.make();
    H2O.submitTask(new RebalanceDataSet(this, f, k)).join();
    Frame f2 = DKV.get(k).get();
    DKV.remove(k);
    return f2;
  }

 /** Appends a named column, keeping the last Vec as the response */
  public Frame add( String name, Vec vec ) {
    if( find(name) != -1 ) throw new IllegalArgumentException("Duplicate name '"+name+"' in Frame");
    if( _vecs.length != 0 ) {
      if( !anyVec().group().equals(vec.group()) && !Arrays.equals(anyVec()._espc,vec._espc) )
        throw new IllegalArgumentException("Vector groups differs - adding vec '"+name+"' into the frame " + Arrays.toString(_names));
      if( numRows() != vec.length() )
        throw new IllegalArgumentException("Vector lengths differ - adding vec '"+name+"' into the frame " + Arrays.toString(_names));
    }
    final int len = _names != null ? _names.length : 0;
    _names = _names != null ? Arrays.copyOf(_names,len+1) : new String[len+1];
    _vecs  = _names != null ? Arrays.copyOf(_vecs ,len+1) : new Vec   [len+1];
    _keys  = _names != null ? Arrays.copyOf(_keys ,len+1) : new Key   [len+1];
    _names[len] = name;
{'fraction_non_alphanumeric': 0.12257189365165491, 'fraction_numerical': 0.006728160607704829, 'mean_word_length': 2.8793938118290887, 'pattern_counts': {'":': 0, '<': 53, '<?xml version=': 0, '>': 35, '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': '6701520', 'n_tokens_mistral': 5858, 'n_tokens_neox': 5741, 'n_words': 2099}