<commit_before>package water.parser;

import java.io.IOException;
import java.io.InputStream;
import java.util.Arrays;
import java.util.HashSet;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.zip.*;
import jsr166y.CountedCompleter;
import water.*;
import water.parser.Parser.ColType;
import water.parser.Parser.ColTypeInfo;
import water.fvec.*;
import water.fvec.Vec.VectorGroup;
import water.nbhm.NonBlockingHashMap;
import water.nbhm.NonBlockingSetInt;
import water.util.ArrayUtils;
import water.util.FrameUtils;
import water.util.PrettyPrint;
import water.util.Log;

public final class ParseDataset extends Job<Frame> {
  private MultiFileParseTask _mfpt; // Access to partially built vectors for cleanup after parser crash

  // Keys are limited to ByteVec Keys and Frames-of-1-ByteVec Keys
  public static Frame parse(Key okey, Key... keys) { return parse(okey,keys,true, false,0/*guess header*/); }

  // Guess setup from inspecting the first Key only, then parse.
  // Suitable for e.g. testing setups, where the data is known to be sane.
  // NOT suitable for random user input!
  public static Frame parse(Key okey, Key[] keys, boolean delete_on_done, boolean singleQuote, int checkHeader) {
    return parse(okey,keys,delete_on_done,setup(keys[0],singleQuote,checkHeader));
  }
  public static Frame parse(Key okey, Key[] keys, boolean delete_on_done, ParseSetup globalSetup) {
    return parse(okey,keys,delete_on_done,globalSetup,true).get();
  }
  public static ParseDataset parse(Key okey, Key[] keys, boolean delete_on_done, ParseSetup globalSetup, boolean blocking) {
    ParseDataset job = forkParseDataset(okey,keys,globalSetup,delete_on_done);
    try { if( blocking ) job.get(); return job; } 
    catch( Throwable ex ) {

      // Took a crash/NPE somewhere in the parser.  Attempt cleanup.
      Futures fs = new Futures();
      if( job != null ) {
        Keyed.remove(job._dest,fs);
        // Find & remove all partially-built output vecs & chunks
        if( job._mfpt != null ) job._mfpt.onExceptionCleanup(fs);
      }
      // Assume the input is corrupt - or already partially deleted after
      // parsing.  Nuke it all - no partial Vecs lying around.
      for( Key k : keys ) Keyed.remove(k,fs);
      fs.blockForPending();
      assert DKV.<Job>getGet(job._key).isStopped();
      throw ex;
    }
  }

  private static ParseSetup setup(Key k, boolean singleQuote, int checkHeader) {
    byte[] bits = ZipUtil.getFirstUnzippedBytes(getByteVec(k));
    ParseSetup globalSetup = ParseSetup.guessSetup(bits, singleQuote, checkHeader);
    if( globalSetup._ncols <= 0 ) throw new UnsupportedOperationException(globalSetup.toString());
    return globalSetup;
  }

  // Allow both ByteVec keys and Frame-of-1-ByteVec
  static ByteVec getByteVec(Key key) {
    Iced ice = DKV.get(key).get();
    return (ByteVec)(ice instanceof ByteVec ? ice : ((Frame)ice).vecs()[0]);
  }
  static String [] genericColumnNames(int ncols){
    String [] res = new String[ncols];
    for(int i = 0; i < res.length; ++i) res[i] = "C" + String.valueOf(i+1);
    return res;
  }

  // Same parse, as a backgroundable Job
  public static ParseDataset forkParseDataset(final Key dest, final Key[] keys, final ParseSetup setup, boolean delete_on_done) {
    HashSet<String> conflictingNames = setup.checkDupColumnNames();
    for( String x : conflictingNames )
      throw new IllegalArgumentException("Found duplicate column name "+x);
    // Some quick sanity checks: no overwriting your input key, and a resource check.
    long totalParseSize=0;
    ByteVec bv;
    float dcr, maxDecompRatio = 0;
    for( int i=0; i<keys.length; i++ ) {
      Key k = keys[i];
      if( dest.equals(k) )
        throw new IllegalArgumentException("Destination key "+dest+" must be different from all sources");
      if( delete_on_done )
        for( int j=i+1; j<keys.length; j++ )
          if( k==keys[j] )
            throw new IllegalArgumentException("Source key "+k+" appears twice, delete_on_done must be false");

      // estimate total size in bytes
      bv = getByteVec(k);
      dcr = ZipUtil.decompressionRatio(bv);
      if (dcr > maxDecompRatio) maxDecompRatio = dcr;
      totalParseSize += bv.length() * maxDecompRatio; // Sum of all input filesizes
    }
    //Calc chunk-size
    setup._chunkSize = Vec.calcOptimalChunkSize(totalParseSize, setup._ncols);
    Log.info("Chunk size " + setup._chunkSize);
    Vec update;
    Iced ice;
    for( int i = 0; i < keys.length; ++i ) {
      ice = DKV.getGet(keys[i]);
      update = (Vec)(ice instanceof Vec ? ice : ((Frame)ice).vec(0));
      //update = ((Frame) DKV.getGet(keys[i])).vec(0);
      update.setChunkSize(setup._chunkSize);
      DKV.put(update._key, update);
    }

    long memsz = H2O.CLOUD.memsz();
    if( totalParseSize > memsz*4 )
      throw new IllegalArgumentException("Total input file size of "+PrettyPrint.bytes(totalParseSize)+" is much larger than total cluster memory of "+PrettyPrint.bytes(memsz)+", please use either a larger cluster or smaller data.");

    // Fire off the parse
    ParseDataset job = new ParseDataset(dest);
    new Frame(job.dest(),new String[0],new Vec[0]).delete_and_lock(job._key); // Write-Lock BEFORE returning
    for( Key k : keys ) Lockable.read_lock(k,job._key); // Read-Lock BEFORE returning
    ParserFJTask fjt = new ParserFJTask(job, keys, setup, delete_on_done); // Fire off background parse
    job.start(fjt, totalParseSize);
    return job;
  }<commit_msg>Fixes break in UploadFiles caused by latest parse changes.  UploadFileVecs use DFLT_CHUNK_SIZE since they don't know the file size until it is done loading.  This is already done before forkParseDataset() sees the data.  This catches this case and leaves the chunkSize as is in these cases.<commit_after>package water.parser;

import java.io.IOException;
import java.io.InputStream;
import java.util.Arrays;
import java.util.HashSet;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.zip.*;
import jsr166y.CountedCompleter;
import water.*;
import water.parser.Parser.ColType;
import water.parser.Parser.ColTypeInfo;
import water.fvec.*;
import water.fvec.Vec.VectorGroup;
import water.nbhm.NonBlockingHashMap;
import water.nbhm.NonBlockingSetInt;
import water.util.ArrayUtils;
import water.util.FrameUtils;
import water.util.PrettyPrint;
import water.util.Log;

public final class ParseDataset extends Job<Frame> {
  private MultiFileParseTask _mfpt; // Access to partially built vectors for cleanup after parser crash

  // Keys are limited to ByteVec Keys and Frames-of-1-ByteVec Keys
  public static Frame parse(Key okey, Key... keys) { return parse(okey,keys,true, false,0/*guess header*/); }

  // Guess setup from inspecting the first Key only, then parse.
  // Suitable for e.g. testing setups, where the data is known to be sane.
  // NOT suitable for random user input!
  public static Frame parse(Key okey, Key[] keys, boolean delete_on_done, boolean singleQuote, int checkHeader) {
    return parse(okey,keys,delete_on_done,setup(keys[0],singleQuote,checkHeader));
  }
  public static Frame parse(Key okey, Key[] keys, boolean delete_on_done, ParseSetup globalSetup) {
    return parse(okey,keys,delete_on_done,globalSetup,true).get();
  }
  public static ParseDataset parse(Key okey, Key[] keys, boolean delete_on_done, ParseSetup globalSetup, boolean blocking) {
    ParseDataset job = forkParseDataset(okey,keys,globalSetup,delete_on_done);
    try { if( blocking ) job.get(); return job; } 
    catch( Throwable ex ) {

      // Took a crash/NPE somewhere in the parser.  Attempt cleanup.
      Futures fs = new Futures();
      if( job != null ) {
        Keyed.remove(job._dest,fs);
        // Find & remove all partially-built output vecs & chunks
        if( job._mfpt != null ) job._mfpt.onExceptionCleanup(fs);
      }
      // Assume the input is corrupt - or already partially deleted after
      // parsing.  Nuke it all - no partial Vecs lying around.
      for( Key k : keys ) Keyed.remove(k,fs);
      fs.blockForPending();
      assert DKV.<Job>getGet(job._key).isStopped();
      throw ex;
    }
  }

  private static ParseSetup setup(Key k, boolean singleQuote, int checkHeader) {
    byte[] bits = ZipUtil.getFirstUnzippedBytes(getByteVec(k));
    ParseSetup globalSetup = ParseSetup.guessSetup(bits, singleQuote, checkHeader);
    if( globalSetup._ncols <= 0 ) throw new UnsupportedOperationException(globalSetup.toString());
    return globalSetup;
  }

  // Allow both ByteVec keys and Frame-of-1-ByteVec
  static ByteVec getByteVec(Key key) {
    Iced ice = DKV.get(key).get();
    return (ByteVec)(ice instanceof ByteVec ? ice : ((Frame)ice).vecs()[0]);
  }
  static String [] genericColumnNames(int ncols){
    String [] res = new String[ncols];
    for(int i = 0; i < res.length; ++i) res[i] = "C" + String.valueOf(i+1);
    return res;
  }

  // Same parse, as a backgroundable Job
  public static ParseDataset forkParseDataset(final Key dest, final Key[] keys, final ParseSetup setup, boolean delete_on_done) {
    HashSet<String> conflictingNames = setup.checkDupColumnNames();
    for( String x : conflictingNames )
      throw new IllegalArgumentException("Found duplicate column name "+x);
    // Some quick sanity checks: no overwriting your input key, and a resource check.
    long totalParseSize=0;
    ByteVec bv;
    float dcr, maxDecompRatio = 0;
    for( int i=0; i<keys.length; i++ ) {
      Key k = keys[i];
      if( dest.equals(k) )
        throw new IllegalArgumentException("Destination key "+dest+" must be different from all sources");
      if( delete_on_done )
        for( int j=i+1; j<keys.length; j++ )
          if( k==keys[j] )
            throw new IllegalArgumentException("Source key "+k+" appears twice, delete_on_done must be false");

      // estimate total size in bytes
      bv = getByteVec(k);
      dcr = ZipUtil.decompressionRatio(bv);
      if (dcr > maxDecompRatio) maxDecompRatio = dcr;
      totalParseSize += bv.length() * maxDecompRatio; // Sum of all input filesizes
    }
    //Calc chunk-size
    //
    Iced ice = DKV.getGet(keys[0]);
    if (ice instanceof Frame && ((Frame) ice).vec(0) instanceof UploadFileVec) {
      setup._chunkSize = Vec.DFLT_CHUNK_SIZE;
    } else {
      setup._chunkSize = Vec.calcOptimalChunkSize(totalParseSize, setup._ncols);
    }
    Log.info("Chunk size " + setup._chunkSize);
    Vec update;
    for( int i = 0; i < keys.length; ++i ) {
      ice = DKV.getGet(keys[i]);
      update = (Vec)(ice instanceof Vec ? ice : ((Frame)ice).vec(0));
      update.setChunkSize(setup._chunkSize);
      DKV.put(update._key, update);
    }

    long memsz = H2O.CLOUD.memsz();
    if( totalParseSize > memsz*4 )
      throw new IllegalArgumentException("Total input file size of "+PrettyPrint.bytes(totalParseSize)+" is much larger than total cluster memory of "+PrettyPrint.bytes(memsz)+", please use either a larger cluster or smaller data.");

    // Fire off the parse
    ParseDataset job = new ParseDataset(dest);
    new Frame(job.dest(),new String[0],new Vec[0]).delete_and_lock(job._key); // Write-Lock BEFORE returning
    for( Key k : keys ) Lockable.read_lock(k,job._key); // Read-Lock BEFORE returning
    ParserFJTask fjt = new ParserFJTask(job, keys, setup, delete_on_done); // Fire off background parse
    job.start(fjt, totalParseSize);
    return job;
  }
{'fraction_non_alphanumeric': 0.11226357535082367, 'fraction_numerical': 0.0039222522443998955, 'mean_word_length': 3.628479225494151, 'pattern_counts': {'":': 0, '<': 19, '<?xml version=': 0, '>': 13, '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': '6811381', 'n_tokens_mistral': 3541, 'n_tokens_neox': 3493, 'n_words': 1058}