gt:                 charSeeker.seek( mark, new int[]{columnDelimiter} );
context: package com.ldbc.driver.workloads.ldbc.snb.interactive;
  
  import com.fasterxml.jackson.core.type.TypeReference;
  import com.fasterxml.jackson.databind.ObjectMapper;
  import com.google.common.base.Charsets;
  import com.google.common.base.Function;
  import com.google.common.base.Predicate;
  import com.google.common.collect.Iterables;
  import com.google.common.collect.Iterators;
  import com.google.common.collect.Lists;
  import com.google.common.collect.PeekingIterator;
  import com.google.common.collect.Sets;
  import com.ldbc.driver.ChildOperationGenerator;
  import com.ldbc.driver.Operation;
  import com.ldbc.driver.SerializingMarshallingException;
  import com.ldbc.driver.Workload;
  import com.ldbc.driver.WorkloadException;
  import com.ldbc.driver.WorkloadStreams;
  import com.ldbc.driver.control.ConsoleAndFileDriverConfiguration;
  import com.ldbc.driver.csv.charseeker.BufferedCharSeeker;
  import com.ldbc.driver.csv.charseeker.CharSeeker;
  import com.ldbc.driver.csv.charseeker.Extractors;
  import com.ldbc.driver.csv.charseeker.Mark;
  import com.ldbc.driver.csv.charseeker.Readables;
  import com.ldbc.driver.csv.charseeker.ThreadAheadReadable;
  import com.ldbc.driver.csv.simple.SimpleCsvFileReader;
  import com.ldbc.driver.generator.CsvEventStreamReaderBasicCharSeeker;
  import com.ldbc.driver.generator.GeneratorFactory;
  import com.ldbc.driver.generator.RandomDataGeneratorFactory;
  import com.ldbc.driver.util.ClassLoaderHelper;
  import com.ldbc.driver.util.ClassLoadingException;
  import com.ldbc.driver.util.Tuple;
  import com.ldbc.driver.util.Tuple2;
  import org.apache.commons.collections4.CollectionUtils;
  import org.apache.commons.collections4.Equator;
  
  import java.io.Closeable;
  import java.io.File;
  import java.io.FileInputStream;
  import java.io.FileNotFoundException;
  import java.io.IOException;
  import java.io.InputStreamReader;
  import java.util.ArrayList;
  import java.util.Collections;
  import java.util.Date;
  import java.util.HashMap;
  import java.util.HashSet;
  import java.util.Iterator;
  import java.util.List;
  import java.util.Map;
  import java.util.NoSuchElementException;
  import java.util.Queue;
  import java.util.Set;
  import java.util.concurrent.TimeUnit;
  
  import static java.lang.String.format;
  
  public class LdbcSnbInteractiveWorkload extends Workload
  {
      private List<Closeable> forumUpdateOperationsFileReaders = new ArrayList<>();
      private List<File> forumUpdateOperationFiles = new ArrayList<>();
      private List<Closeable> personUpdateOperationsFileReaders = new ArrayList<>();
      private List<File> personUpdateOperationFiles = new ArrayList<>();
  
      private List<Closeable> readOperationFileReaders = new ArrayList<>();
      private File readOperation1File;
      private File readOperation2File;
      private File readOperation3File;
      private File readOperation4File;
      private File readOperation5File;
      private File readOperation6File;
      private File readOperation7File;
      private File readOperation8File;
      private File readOperation9File;
      private File readOperation10File;
      private File readOperation11File;
      private File readOperation12File;
      private File readOperation13File;
      private File readOperation14File;
  
      private long readOperation1InterleaveAsMilli;
      private long readOperation2InterleaveAsMilli;
      private long readOperation3InterleaveAsMilli;
      private long readOperation4InterleaveAsMilli;
      private long readOperation5InterleaveAsMilli;
      private long readOperation6InterleaveAsMilli;
      private long readOperation7InterleaveAsMilli;
      private long readOperation8InterleaveAsMilli;
      private long readOperation9InterleaveAsMilli;
      private long readOperation10InterleaveAsMilli;
      private long readOperation11InterleaveAsMilli;
      private long readOperation12InterleaveAsMilli;
      private long readOperation13InterleaveAsMilli;
      private long readOperation14InterleaveAsMilli;
  
      private long updateInterleaveAsMilli;
      private double compressionRatio;
      private double shortReadDissipationFactor;
  
      private Set<Class> enabledLongReadOperationTypes;
      private Set<Class> enabledShortReadOperationTypes;
      private Set<Class> enabledWriteOperationTypes;
      private LdbcSnbInteractiveWorkloadConfiguration.UpdateStreamParser parser;
  
      @Override
      public Map<Integer,Class<? extends Operation>> operationTypeToClassMapping()
      {
          return LdbcSnbInteractiveWorkloadConfiguration.operationTypeToClassMapping();
      }
  
      @Override
      public void onInit( Map<String,String> params ) throws WorkloadException
      {
          List<String> compulsoryKeys = Lists.newArrayList(
                  LdbcSnbInteractiveWorkloadConfiguration.PARAMETERS_DIRECTORY );
  
          compulsoryKeys.addAll( LdbcSnbInteractiveWorkloadConfiguration.LONG_READ_OPERATION_ENABLE_KEYS );
          compulsoryKeys.addAll( LdbcSnbInteractiveWorkloadConfiguration.WRITE_OPERATION_ENABLE_KEYS );
          compulsoryKeys.addAll( LdbcSnbInteractiveWorkloadConfiguration.SHORT_READ_OPERATION_ENABLE_KEYS );
  
          Set<String> missingPropertyParameters =
                  LdbcSnbInteractiveWorkloadConfiguration.missingParameters( params, compulsoryKeys );
          if ( false == missingPropertyParameters.isEmpty() )
          {
              throw new WorkloadException( format( "Workload could not initialize due to missing parameters: %s",
                      missingPropertyParameters.toString() ) );
          }
  
          if ( params.containsKey( LdbcSnbInteractiveWorkloadConfiguration.UPDATES_DIRECTORY ) )
          {
              String updatesDirectoryPath =
                      params.get( LdbcSnbInteractiveWorkloadConfiguration.UPDATES_DIRECTORY ).trim();
              File updatesDirectory = new File( updatesDirectoryPath );
              if ( false == updatesDirectory.exists() )
              {
                  throw new WorkloadException( format( "Updates directory does not exist\nDirectory: %s",
                          updatesDirectory.getAbsolutePath() ) );
              }
              if ( false == updatesDirectory.isDirectory() )
              {
                  throw new WorkloadException( format( "Updates directory is not a directory\nDirectory: %s",
                          updatesDirectory.getAbsolutePath() ) );
              }
              forumUpdateOperationFiles = LdbcSnbInteractiveWorkloadConfiguration
                      .forumUpdateFilesInDirectory( updatesDirectory );
              personUpdateOperationFiles =
                      LdbcSnbInteractiveWorkloadConfiguration.personUpdateFilesInDirectory( updatesDirectory );
          }
          else
          {
              forumUpdateOperationFiles = new ArrayList<>();
              personUpdateOperationFiles = new ArrayList<>();
          }
  
          File parametersDir =
                  new File( params.get( LdbcSnbInteractiveWorkloadConfiguration.PARAMETERS_DIRECTORY ).trim() );
          if ( false == parametersDir.exists() )
          {
              throw new WorkloadException(
                      format( "Parameters directory does not exist: %s", parametersDir.getAbsolutePath() ) );
          }
          for ( String readOperationParamsFilename :
                  LdbcSnbInteractiveWorkloadConfiguration.READ_OPERATION_PARAMS_FILENAMES )
          {
              File readOperationParamsFile = new File( parametersDir, readOperationParamsFilename );
              if ( false == readOperationParamsFile.exists() )
              {
                  throw new WorkloadException(
                          format( "Read operation parameters file does not exist: %s",
                                  readOperationParamsFile.getAbsolutePath() ) );
              }
          }
          readOperation1File =
                  new File( parametersDir, LdbcSnbInteractiveWorkloadConfiguration.READ_OPERATION_1_PARAMS_FILENAME );
          readOperation2File =
                  new File( parametersDir, LdbcSnbInteractiveWorkloadConfiguration.READ_OPERATION_2_PARAMS_FILENAME );
          readOperation3File =
                  new File( parametersDir, LdbcSnbInteractiveWorkloadConfiguration.READ_OPERATION_3_PARAMS_FILENAME );
          readOperation4File =
                  new File( parametersDir, LdbcSnbInteractiveWorkloadConfiguration.READ_OPERATION_4_PARAMS_FILENAME );
          readOperation5File =
                  new File( parametersDir, LdbcSnbInteractiveWorkloadConfiguration.READ_OPERATION_5_PARAMS_FILENAME );
          readOperation7File =
                  new File( parametersDir, LdbcSnbInteractiveWorkloadConfiguration.READ_OPERATION_7_PARAMS_FILENAME );
          readOperation8File =
                  new File( parametersDir, LdbcSnbInteractiveWorkloadConfiguration.READ_OPERATION_8_PARAMS_FILENAME );
          readOperation9File =
                  new File( parametersDir, LdbcSnbInteractiveWorkloadConfiguration.READ_OPERATION_9_PARAMS_FILENAME );
          readOperation6File =
                  new File( parametersDir, LdbcSnbInteractiveWorkloadConfiguration.READ_OPERATION_6_PARAMS_FILENAME );
          readOperation10File =
                  new File( parametersDir, LdbcSnbInteractiveWorkloadConfiguration.READ_OPERATION_10_PARAMS_FILENAME );
          readOperation11File =
                  new File( parametersDir, LdbcSnbInteractiveWorkloadConfiguration.READ_OPERATION_11_PARAMS_FILENAME );
          readOperation12File =
                  new File( parametersDir, LdbcSnbInteractiveWorkloadConfiguration.READ_OPERATION_12_PARAMS_FILENAME );
          readOperation13File =
                  new File( parametersDir, LdbcSnbInteractiveWorkloadConfiguration.READ_OPERATION_13_PARAMS_FILENAME );
          readOperation14File =
                  new File( parametersDir, LdbcSnbInteractiveWorkloadConfiguration.READ_OPERATION_14_PARAMS_FILENAME );
  
          enabledLongReadOperationTypes = new HashSet<>();
          for ( String longReadOperationEnableKey : LdbcSnbInteractiveWorkloadConfiguration
                  .LONG_READ_OPERATION_ENABLE_KEYS )
          {
              String longReadOperationEnabledString = params.get( longReadOperationEnableKey ).trim();
              Boolean longReadOperationEnabled = Boolean.parseBoolean( longReadOperationEnabledString );
              String longReadOperationClassName =
                      LdbcSnbInteractiveWorkloadConfiguration.LDBC_INTERACTIVE_PACKAGE_PREFIX +
                      LdbcSnbInteractiveWorkloadConfiguration.removePrefix(
                              LdbcSnbInteractiveWorkloadConfiguration.removeSuffix(
                                      longReadOperationEnableKey,
                                      LdbcSnbInteractiveWorkloadConfiguration.ENABLE_SUFFIX
                              ),
                              LdbcSnbInteractiveWorkloadConfiguration
                                      .LDBC_SNB_INTERACTIVE_PARAM_NAME_PREFIX
                      );
              try
              {
                  Class longReadOperationClass = ClassLoaderHelper.loadClass( longReadOperationClassName );
                  if ( longReadOperationEnabled )
                  { enabledLongReadOperationTypes.add( longReadOperationClass ); }
              }
              catch ( ClassLoadingException e )
              {
                  throw new WorkloadException(
                          format(
                                  "Unable to load operation class for parameter: %s\nGuessed incorrect class name: %s",
                                  longReadOperationEnableKey, longReadOperationClassName ),
                          e
                  );
              }
          }
  
          enabledShortReadOperationTypes = new HashSet<>();
          for ( String shortReadOperationEnableKey : LdbcSnbInteractiveWorkloadConfiguration
                  .SHORT_READ_OPERATION_ENABLE_KEYS )
          {
              String shortReadOperationEnabledString = params.get( shortReadOperationEnableKey ).trim();
              Boolean shortReadOperationEnabled = Boolean.parseBoolean( shortReadOperationEnabledString );
              String shortReadOperationClassName =
                      LdbcSnbInteractiveWorkloadConfiguration.LDBC_INTERACTIVE_PACKAGE_PREFIX +
                      LdbcSnbInteractiveWorkloadConfiguration.removePrefix(
                              LdbcSnbInteractiveWorkloadConfiguration.removeSuffix(
                                      shortReadOperationEnableKey,
                                      LdbcSnbInteractiveWorkloadConfiguration.ENABLE_SUFFIX
                              ),
                              LdbcSnbInteractiveWorkloadConfiguration
                                      .LDBC_SNB_INTERACTIVE_PARAM_NAME_PREFIX
                      );
              try
              {
                  Class shortReadOperationClass = ClassLoaderHelper.loadClass( shortReadOperationClassName );
                  if ( shortReadOperationEnabled )
                  { enabledShortReadOperationTypes.add( shortReadOperationClass ); }
              }
              catch ( ClassLoadingException e )
              {
                  throw new WorkloadException(
                          format(
                                  "Unable to load operation class for parameter: %s\nGuessed incorrect class name: %s",
                                  shortReadOperationEnableKey, shortReadOperationClassName ),
                          e
                  );
              }
          }
          if ( false == enabledShortReadOperationTypes.isEmpty() )
          {
              if ( false == params.containsKey( LdbcSnbInteractiveWorkloadConfiguration.SHORT_READ_DISSIPATION ) )
              {
                  throw new WorkloadException( format( "Configuration parameter missing: %s",
                          LdbcSnbInteractiveWorkloadConfiguration.SHORT_READ_DISSIPATION ) );
              }
              shortReadDissipationFactor = Double.parseDouble(
                      params.get( LdbcSnbInteractiveWorkloadConfiguration.SHORT_READ_DISSIPATION ).trim()
              );
              if ( shortReadDissipationFactor < 0 || shortReadDissipationFactor > 1 )
              {
                  throw new WorkloadException(
                          format( "Configuration parameter %s should be in interval [1.0,0.0] but is: %s",
                                  shortReadDissipationFactor ) );
              }
          }
  
          enabledWriteOperationTypes = new HashSet<>();
          for ( String writeOperationEnableKey : LdbcSnbInteractiveWorkloadConfiguration.WRITE_OPERATION_ENABLE_KEYS )
          {
              String writeOperationEnabledString = params.get( writeOperationEnableKey ).trim();
              Boolean writeOperationEnabled = Boolean.parseBoolean( writeOperationEnabledString );
              String writeOperationClassName = LdbcSnbInteractiveWorkloadConfiguration.LDBC_INTERACTIVE_PACKAGE_PREFIX +
                                               LdbcSnbInteractiveWorkloadConfiguration.removePrefix(
                                                       LdbcSnbInteractiveWorkloadConfiguration.removeSuffix(
                                                               writeOperationEnableKey,
                                                               LdbcSnbInteractiveWorkloadConfiguration.ENABLE_SUFFIX
                                                       ),
                                                       LdbcSnbInteractiveWorkloadConfiguration
                                                               .LDBC_SNB_INTERACTIVE_PARAM_NAME_PREFIX
                                               );
              try
              {
                  Class writeOperationClass = ClassLoaderHelper.loadClass( writeOperationClassName );
                  if ( writeOperationEnabled )
                  { enabledWriteOperationTypes.add( writeOperationClass ); }
              }
              catch ( ClassLoadingException e )
              {
                  throw new WorkloadException(
                          format(
                                  "Unable to load operation class for parameter: %s\nGuessed incorrect class name: %s",
                                  writeOperationEnableKey, writeOperationClassName ),
                          e
                  );
              }
          }
  
          List<String> frequencyKeys =
                  Lists.newArrayList( LdbcSnbInteractiveWorkloadConfiguration.READ_OPERATION_FREQUENCY_KEYS );
          Set<String> missingFrequencyKeys = LdbcSnbInteractiveWorkloadConfiguration
                  .missingParameters( params, frequencyKeys );
          if ( enabledWriteOperationTypes.isEmpty() &&
               false == params.containsKey( LdbcSnbInteractiveWorkloadConfiguration.UPDATE_INTERLEAVE ) )
          {
              // if UPDATE_INTERLEAVE is missing and writes are disabled set it to DEFAULT
              params.put(
                      LdbcSnbInteractiveWorkloadConfiguration.UPDATE_INTERLEAVE,
                      LdbcSnbInteractiveWorkloadConfiguration.DEFAULT_UPDATE_INTERLEAVE
              );
          }
          if ( false == params.containsKey( LdbcSnbInteractiveWorkloadConfiguration.UPDATE_INTERLEAVE ) )
          {
              // if UPDATE_INTERLEAVE is missing but writes are enabled it is an error
              throw new WorkloadException(
                      format( "Workload could not initialize. Missing parameter: %s",
                              LdbcSnbInteractiveWorkloadConfiguration.UPDATE_INTERLEAVE ) );
          }
          updateInterleaveAsMilli =
                  Integer.parseInt( params.get( LdbcSnbInteractiveWorkloadConfiguration.UPDATE_INTERLEAVE ).trim() );
          if ( missingFrequencyKeys.isEmpty() )
          {
              // all frequency arguments were given, compute interleave based on frequencies
              params = LdbcSnbInteractiveWorkloadConfiguration.convertFrequenciesToInterleaves( params );
          }
          else
          {
              // if any frequencies are not set, there should be specified interleave times for read queries
              Set<String> missingInterleaveKeys = LdbcSnbInteractiveWorkloadConfiguration.missingParameters(
                      params,
                      LdbcSnbInteractiveWorkloadConfiguration.READ_OPERATION_INTERLEAVE_KEYS
              );
              if ( false == missingInterleaveKeys.isEmpty() )
              {
                  throw new WorkloadException( format(
                          "Workload could not initialize. One of the following groups of parameters should be set: %s " +
                          "or %s",
                          missingFrequencyKeys.toString(), missingInterleaveKeys.toString() ) );
              }
          }
  
          try
          {
              readOperation1InterleaveAsMilli = Long.parseLong(
                      params.get( LdbcSnbInteractiveWorkloadConfiguration.READ_OPERATION_1_INTERLEAVE_KEY ).trim() );
              readOperation2InterleaveAsMilli = Long.parseLong(
                      params.get( LdbcSnbInteractiveWorkloadConfiguration.READ_OPERATION_2_INTERLEAVE_KEY ).trim() );
              readOperation3InterleaveAsMilli = Long.parseLong(
                      params.get( LdbcSnbInteractiveWorkloadConfiguration.READ_OPERATION_3_INTERLEAVE_KEY ).trim() );
              readOperation4InterleaveAsMilli = Long.parseLong(
                      params.get( LdbcSnbInteractiveWorkloadConfiguration.READ_OPERATION_4_INTERLEAVE_KEY ).trim() );
              readOperation5InterleaveAsMilli = Long.parseLong(
                      params.get( LdbcSnbInteractiveWorkloadConfiguration.READ_OPERATION_5_INTERLEAVE_KEY ).trim() );
              readOperation6InterleaveAsMilli = Long.parseLong(
                      params.get( LdbcSnbInteractiveWorkloadConfiguration.READ_OPERATION_6_INTERLEAVE_KEY ).trim() );
              readOperation7InterleaveAsMilli = Long.parseLong(
                      params.get( LdbcSnbInteractiveWorkloadConfiguration.READ_OPERATION_7_INTERLEAVE_KEY ).trim() );
              readOperation8InterleaveAsMilli = Long.parseLong(
                      params.get( LdbcSnbInteractiveWorkloadConfiguration.READ_OPERATION_8_INTERLEAVE_KEY ).trim() );
              readOperation9InterleaveAsMilli = Long.parseLong(
                      params.get( LdbcSnbInteractiveWorkloadConfiguration.READ_OPERATION_9_INTERLEAVE_KEY ).trim() );
              readOperation10InterleaveAsMilli = Long.parseLong(
                      params.get( LdbcSnbInteractiveWorkloadConfiguration.READ_OPERATION_10_INTERLEAVE_KEY ).trim() );
              readOperation11InterleaveAsMilli = Long.parseLong(
                      params.get( LdbcSnbInteractiveWorkloadConfiguration.READ_OPERATION_11_INTERLEAVE_KEY ).trim() );
              readOperation12InterleaveAsMilli = Long.parseLong(
                      params.get( LdbcSnbInteractiveWorkloadConfiguration.READ_OPERATION_12_INTERLEAVE_KEY ).trim() );
              readOperation13InterleaveAsMilli = Long.parseLong(
                      params.get( LdbcSnbInteractiveWorkloadConfiguration.READ_OPERATION_13_INTERLEAVE_KEY ).trim() );
              readOperation14InterleaveAsMilli = Long.parseLong(
                      params.get( LdbcSnbInteractiveWorkloadConfiguration.READ_OPERATION_14_INTERLEAVE_KEY ).trim() );
          }
          catch ( NumberFormatException e )
          {
              throw new WorkloadException( "Unable to parse one of the read operation interleave values", e );
          }
  
          String parserString = params.get( LdbcSnbInteractiveWorkloadConfiguration.UPDATE_STREAM_PARSER );
          if ( null == parserString )
          {
              parserString = LdbcSnbInteractiveWorkloadConfiguration.DEFAULT_UPDATE_STREAM_PARSER.name();
          }
          if ( false == LdbcSnbInteractiveWorkloadConfiguration.isValidParser( parserString ) )
          {
              throw new WorkloadException( "Invalid parser: " + parserString );
          }
          this.parser = LdbcSnbInteractiveWorkloadConfiguration.UpdateStreamParser.valueOf( parserString.trim() );
          this.compressionRatio = Double.parseDouble(
                  params.get( ConsoleAndFileDriverConfiguration.TIME_COMPRESSION_RATIO_ARG ).trim()
          );
      }
  
      @Override
      synchronized protected void onClose() throws IOException
      {
          for ( Closeable forumUpdateOperationsFileReader : forumUpdateOperationsFileReaders )
          {
              forumUpdateOperationsFileReader.close();
          }
  
          for ( Closeable personUpdateOperationsFileReader : personUpdateOperationsFileReaders )
          {
              personUpdateOperationsFileReader.close();
          }
  
          for ( Closeable readOperationFileReader : readOperationFileReaders )
          {
              readOperationFileReader.close();
          }
      }
  
      private Tuple2<Iterator<Operation>,Closeable> fileToWriteStreamParser( File updateOperationsFile,
              LdbcSnbInteractiveWorkloadConfiguration.UpdateStreamParser parser ) throws IOException, WorkloadException
      {
          switch ( parser )
          {
          case REGEX:
          {
              SimpleCsvFileReader csvFileReader = new SimpleCsvFileReader( updateOperationsFile,
                      SimpleCsvFileReader.DEFAULT_COLUMN_SEPARATOR_REGEX_STRING );
              return Tuple.<Iterator<Operation>,Closeable>tuple2( WriteEventStreamReaderRegex.create( csvFileReader ),
                      csvFileReader );
          }
          case CHAR_SEEKER:
          {
              int bufferSize = 1 * 1024 * 1024;
  //                BufferedCharSeeker charSeeker = new BufferedCharSeeker(Readables.wrap(new FileReader
  // (updateOperationsFile)), bufferSize);
              BufferedCharSeeker charSeeker = new BufferedCharSeeker(
                      Readables.wrap(
                              new InputStreamReader( new FileInputStream( updateOperationsFile ), Charsets.UTF_8 )
                      ),
                      bufferSize
              );
              Extractors extractors = new Extractors( ';', ',' );
              return Tuple.<Iterator<Operation>,Closeable>tuple2(
                      WriteEventStreamReaderCharSeeker.create( charSeeker, extractors, '|' ), charSeeker );
          }
          case CHAR_SEEKER_THREAD:
          {
              int bufferSize = 1 * 1024 * 1024;
              BufferedCharSeeker charSeeker = new BufferedCharSeeker(
                      ThreadAheadReadable.threadAhead(
                              Readables.wrap(
                                      new InputStreamReader( new FileInputStream( updateOperationsFile ), Charsets.UTF_8 )
                              ),
                              bufferSize
                      ),
                      bufferSize
              );
              Extractors extractors = new Extractors( ';', ',' );
              return Tuple.<Iterator<Operation>,Closeable>tuple2(
                      WriteEventStreamReaderCharSeeker.create( charSeeker, extractors, '|' ), charSeeker );
          }
          }
          SimpleCsvFileReader csvFileReader = new SimpleCsvFileReader( updateOperationsFile,
                  SimpleCsvFileReader.DEFAULT_COLUMN_SEPARATOR_REGEX_STRING );
          return Tuple.<Iterator<Operation>,Closeable>tuple2( WriteEventStreamReaderRegex.create( csvFileReader ),
                  csvFileReader );
      }
  
      @Override
      protected WorkloadStreams getStreams( GeneratorFactory gf, boolean hasDbConnected ) throws WorkloadException
      {
          long workloadStartTimeAsMilli = Long.MAX_VALUE;
          WorkloadStreams ldbcSnbInteractiveWorkloadStreams = new WorkloadStreams();
          List<Iterator<?>> asynchronousDependencyStreamsList = new ArrayList<>();
          List<Iterator<?>> asynchronousNonDependencyStreamsList = new ArrayList<>();
          Set<Class<? extends Operation>> dependentAsynchronousOperationTypes = Sets.newHashSet();
          Set<Class<? extends Operation>> dependencyAsynchronousOperationTypes = Sets.newHashSet();
  
          /* *******
           * *******
           * *******
           *  WRITES
           * *******
           * *******
           * *******/
  
          // TODO put person/forum update stream pairs into same streams, to half required thread count
           /*
           * Create person write operation streams
           */
          if ( enabledWriteOperationTypes.contains( LdbcUpdate1AddPerson.class ) )
          {
              for ( File personUpdateOperationFile : personUpdateOperationFiles )
              {
                  Iterator<Operation> personUpdateOperationsParser;
                  try
                  {
                      Tuple2<Iterator<Operation>,Closeable> parserAndCloseable =
                              fileToWriteStreamParser( personUpdateOperationFile, parser );
                      personUpdateOperationsParser = parserAndCloseable._1();
                      personUpdateOperationsFileReaders.add( parserAndCloseable._2() );
                  }
                  catch ( IOException e )
                  {
                      throw new WorkloadException(
                              "Unable to open person update stream: " + personUpdateOperationFile.getAbsolutePath(), e );
                  }
                  if ( false == personUpdateOperationsParser.hasNext() )
                  {
                      // Update stream is empty
                      System.out.println(
                              format( ""
                                      + "***********************************************\n"
                                      + "  !! WARMING !!\n"
                                      + "  Update stream is empty: %s\n"
                                      + "  Check that data generation process completed successfully\n"
                                      + "***********************************************",
                                      personUpdateOperationFile.getAbsolutePath()
                              )
                      );
                      continue;
                  }
                  PeekingIterator<Operation> unfilteredPersonUpdateOperations =
                          Iterators.peekingIterator( personUpdateOperationsParser );
  
                  try
                  {
                      if ( unfilteredPersonUpdateOperations.peek().scheduledStartTimeAsMilli() <
                           workloadStartTimeAsMilli )
                      {
                          workloadStartTimeAsMilli = unfilteredPersonUpdateOperations.peek().scheduledStartTimeAsMilli();
                      }
                  }
                  catch ( NoSuchElementException e )
                  {
                      // do nothing, exception just means that stream was empty
                  }
  
                  // Filter Write Operations
                  Predicate<Operation> enabledWriteOperationsFilter = new Predicate<Operation>()
                  {
                      @Override
                      public boolean apply( Operation operation )
                      {
                          return enabledWriteOperationTypes.contains( operation.getClass() );
                      }
                  };
                  Iterator<Operation> filteredPersonUpdateOperations =
                          Iterators.filter( unfilteredPersonUpdateOperations, enabledWriteOperationsFilter );
  
                  Set<Class<? extends Operation>> dependentPersonUpdateOperationTypes = Sets.newHashSet();
                  Set<Class<? extends Operation>> dependencyPersonUpdateOperationTypes =
                          Sets.<Class<? extends Operation>>newHashSet(
                                  LdbcUpdate1AddPerson.class
                          );
  
                  ChildOperationGenerator personUpdateChildOperationGenerator = null;
  
                  ldbcSnbInteractiveWorkloadStreams.addBlockingStream(
                          dependentPersonUpdateOperationTypes,
                          dependencyPersonUpdateOperationTypes,
                          filteredPersonUpdateOperations,
                          Collections.<Operation>emptyIterator(),
                          personUpdateChildOperationGenerator
                  );
              }
          }
  
          /*
           * Create forum write operation streams
           */
          if ( enabledWriteOperationTypes.contains( LdbcUpdate2AddPostLike.class ) ||
               enabledWriteOperationTypes.contains( LdbcUpdate3AddCommentLike.class ) ||
               enabledWriteOperationTypes.contains( LdbcUpdate4AddForum.class ) ||
               enabledWriteOperationTypes.contains( LdbcUpdate5AddForumMembership.class ) ||
               enabledWriteOperationTypes.contains( LdbcUpdate6AddPost.class ) ||
               enabledWriteOperationTypes.contains( LdbcUpdate7AddComment.class ) ||
               enabledWriteOperationTypes.contains( LdbcUpdate8AddFriendship.class )
                  )
          {
              for ( File forumUpdateOperationFile : forumUpdateOperationFiles )
              {
                  Iterator<Operation> forumUpdateOperationsParser;
                  try
                  {
                      Tuple2<Iterator<Operation>,Closeable> parserAndCloseable =
                              fileToWriteStreamParser( forumUpdateOperationFile, parser );
                      forumUpdateOperationsParser = parserAndCloseable._1();
                      forumUpdateOperationsFileReaders.add( parserAndCloseable._2() );
                  }
                  catch ( IOException e )
                  {
                      throw new WorkloadException(
                              "Unable to open forum update stream: " + forumUpdateOperationFile.getAbsolutePath(), e );
                  }
                  if ( false == forumUpdateOperationsParser.hasNext() )
                  {
                      // Update stream is empty
                      System.out.println(
                              format( ""
                                      + "***********************************************\n"
                                      + "  !! WARMING !!\n"
                                      + "  Update stream is empty: %s\n"
                                      + "  Check that data generation process completed successfully\n"
                                      + "***********************************************",
                                      forumUpdateOperationFile.getAbsolutePath()
                              )
                      );
                      continue;
                  }
                  PeekingIterator<Operation> unfilteredForumUpdateOperations =
                          Iterators.peekingIterator( forumUpdateOperationsParser );
  
                  try
                  {
                      if ( unfilteredForumUpdateOperations.peek().scheduledStartTimeAsMilli() < workloadStartTimeAsMilli )
                      {
                          workloadStartTimeAsMilli = unfilteredForumUpdateOperations.peek().scheduledStartTimeAsMilli();
                      }
                  }
                  catch ( NoSuchElementException e )
                  {
                      // do nothing, exception just means that stream was empty
                  }
  
                  // Filter Write Operations
                  Predicate<Operation> enabledWriteOperationsFilter = new Predicate<Operation>()
                  {
                      @Override
                      public boolean apply( Operation operation )
                      {
                          return enabledWriteOperationTypes.contains( operation.getClass() );
                      }
                  };
                  Iterator<Operation> filteredForumUpdateOperations =
                          Iterators.filter( unfilteredForumUpdateOperations, enabledWriteOperationsFilter );
  
                  Set<Class<? extends Operation>> dependentForumUpdateOperationTypes =
                          Sets.<Class<? extends Operation>>newHashSet(
                                  LdbcUpdate2AddPostLike.class,
                                  LdbcUpdate3AddCommentLike.class,
                                  LdbcUpdate4AddForum.class,
                                  LdbcUpdate5AddForumMembership.class,
                                  LdbcUpdate6AddPost.class,
                                  LdbcUpdate7AddComment.class,
                                  LdbcUpdate8AddFriendship.class
                          );
                  Set<Class<? extends Operation>> dependencyForumUpdateOperationTypes = Sets.newHashSet();
  
                  ChildOperationGenerator forumUpdateChildOperationGenerator = null;
  
                  ldbcSnbInteractiveWorkloadStreams.addBlockingStream(
                          dependentForumUpdateOperationTypes,
                          dependencyForumUpdateOperationTypes,
                          Collections.<Operation>emptyIterator(),
                          filteredForumUpdateOperations,
                          forumUpdateChildOperationGenerator
                  );
              }
          }
  
          if ( Long.MAX_VALUE == workloadStartTimeAsMilli )
          { workloadStartTimeAsMilli = 0; }
  
          /* *******
           * *******
           * *******
           *  LONG READS
           * *******
           * *******
           * *******/
  
          /*
           * Create read operation streams, with specified interleaves
           */
          int bufferSize = 1 * 1024 * 1024;
          char columnDelimiter = '|';
          char arrayDelimiter = ';';
          char tupleDelimiter = ',';
  
          Iterator<Operation> readOperation1Stream;
          {
              CsvEventStreamReaderBasicCharSeeker.EventDecoder<Object[]> decoder =
                      new Query1EventStreamReader.Query1Decoder();
              Extractors extractors = new Extractors( arrayDelimiter, tupleDelimiter );
              CharSeeker charSeeker;
              try
              {
                  charSeeker = new BufferedCharSeeker(
                          Readables.wrap(
                                  new InputStreamReader( new FileInputStream( readOperation1File ), Charsets.UTF_8 )
                          ),
                          bufferSize
                  );
              }
              catch ( FileNotFoundException e )
              {
                  throw new WorkloadException(
                          format( "Unable to open parameters file: %s", readOperation1File.getAbsolutePath() ),
                          e );
              }
              Mark mark = new Mark();
              // skip headers
              try
              {
                  charSeeker.seek( mark, new int[]{columnDelimiter} );
                  charSeeker.seek( mark, new int[]{columnDelimiter} );
              }
              catch ( IOException e )
              {
                  throw new WorkloadException( format( "Unable to advance parameters file beyond headers: %s",
                          readOperation1File.getAbsolutePath() ), e );
              }
  
              Iterator<Operation> operation1StreamWithoutTimes = new Query1EventStreamReader(
                      gf.repeating(
                              new CsvEventStreamReaderBasicCharSeeker<>(
                                      charSeeker,
                                      extractors,
                                      mark,
                                      decoder,
                                      columnDelimiter
                              )
                      )
              );
  
              Iterator<Long> operation1StartTimes =
                      gf.incrementing( workloadStartTimeAsMilli + readOperation1InterleaveAsMilli,
                              readOperation1InterleaveAsMilli );
  
              readOperation1Stream = gf.assignStartTimes(
                      operation1StartTimes,
                      operation1StreamWithoutTimes
              );
  
              readOperationFileReaders.add( charSeeker );
          }
  
          Iterator<Operation> readOperation2Stream;
          {
              CsvEventStreamReaderBasicCharSeeker.EventDecoder<Object[]> decoder =
                      new Query2EventStreamReader.Query2Decoder();
              Extractors extractors = new Extractors( arrayDelimiter, tupleDelimiter );
              CharSeeker charSeeker;
              try
              {
                  charSeeker = new BufferedCharSeeker(
                          Readables.wrap(
                                  new InputStreamReader( new FileInputStream( readOperation2File ), Charsets.UTF_8 )
                          ),
                          bufferSize
                  );
              }
              catch ( FileNotFoundException e )
              {
                  throw new WorkloadException(
                          format( "Unable to open parameters file: %s", readOperation2File.getAbsolutePath() ),
                          e );
              }
              Mark mark = new Mark();
              // skip headers
              try
              {
                  charSeeker.seek( mark, new int[]{columnDelimiter} );
                  charSeeker.seek( mark, new int[]{columnDelimiter} );
              }
              catch ( IOException e )
              {
                  throw new WorkloadException( format( "Unable to advance parameters file beyond headers: %s",
                          readOperation2File.getAbsolutePath() ), e );
              }
  
              Iterator<Operation> operation2StreamWithoutTimes = new Query2EventStreamReader(
                      gf.repeating(
                              new CsvEventStreamReaderBasicCharSeeker<>(
                                      charSeeker,
                                      extractors,
                                      mark,
                                      decoder,
                                      columnDelimiter
                              )
                      )
              );
  
              Iterator<Long> operation2StartTimes =
                      gf.incrementing( workloadStartTimeAsMilli + readOperation2InterleaveAsMilli,
                              readOperation2InterleaveAsMilli );
  
              readOperation2Stream = gf.assignStartTimes(
                      operation2StartTimes,
                      operation2StreamWithoutTimes
              );
  
              readOperationFileReaders.add( charSeeker );
          }
  
          Iterator<Operation> readOperation3Stream;
          {
              CsvEventStreamReaderBasicCharSeeker.EventDecoder<Object[]> decoder =
                      new Query3EventStreamReader.Query3Decoder();
              Extractors extractors = new Extractors( arrayDelimiter, tupleDelimiter );
              CharSeeker charSeeker;
              try
              {
                  charSeeker = new BufferedCharSeeker(
                          Readables.wrap(
                                  new InputStreamReader( new FileInputStream( readOperation3File ), Charsets.UTF_8 )
                          ),
                          bufferSize
                  );
              }
              catch ( FileNotFoundException e )
              {
                  throw new WorkloadException(
                          format( "Unable to open parameters file: %s", readOperation3File.getAbsolutePath() ),
                          e );
              }
              Mark mark = new Mark();
              // skip headers
              try
              {
                  charSeeker.seek( mark, new int[]{columnDelimiter} );
                  charSeeker.seek( mark, new int[]{columnDelimiter} );
                  charSeeker.seek( mark, new int[]{columnDelimiter} );
                  charSeeker.seek( mark, new int[]{columnDelimiter} );
  