<commit_before>package io.warp10.continuum;

import io.warp10.WarpConfig;
import io.warp10.continuum.gts.GTSDecoder;
import io.warp10.continuum.gts.GTSEncoder;
import io.warp10.continuum.gts.GTSHelper;
import io.warp10.continuum.gts.GTSWrapperHelper;
import io.warp10.continuum.store.Constants;
import io.warp10.continuum.store.thrift.data.GTSWrapper;
import io.warp10.continuum.store.thrift.data.Metadata;
import io.warp10.crypto.OrderPreservingBase64;

import java.io.BufferedReader;
import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map.Entry;
import java.util.Set;

import org.apache.commons.cli.CommandLine;
import org.apache.commons.cli.CommandLineParser;
import org.apache.commons.cli.DefaultParser;
import org.apache.commons.cli.Options;
import org.apache.thrift.TSerializer;
import org.apache.thrift.protocol.TCompactProtocol;
import org.python.jline.internal.InputStreamReader;

/**
 * Packs Geo Time Series into chunks spanning at most a given
 * timespan.
 */
public class Pack {
  
  private static final long MAX_ENCODER_SIZE = 16 * 1024 * 1024L;
  
  // Set of Geo Time Series already encountered
  private static Set<Metadata> parsed = new HashSet<Metadata>();

  private static void pack(long chunksize, int maxsize, String suffix) throws Exception {
    BufferedReader br = new BufferedReader(new InputStreamReader(System.in));
    
    long now = TimeSource.getTime();
        
    GTSEncoder lastencoder = null;
    
    while(true) {
      String line = br.readLine();
      
      if (null == line) {
        break;
      }
      
      GTSEncoder encoder = GTSHelper.parse(lastencoder, line, null, now);

      if (null != lastencoder && (encoder != lastencoder || lastencoder.size() > MAX_ENCODER_SIZE)) {
        // We changed the GTS, pack 'lastencoder'
        outputEncoder(lastencoder, chunksize, maxsize, suffix);
        
        if (encoder == lastencoder) {
          // We allocate a branch new Encoder but set its metadata to those of 'encoder'
          // so continuation lines are supported
          lastencoder = new GTSEncoder(0L);
          lastencoder.setMetadata(encoder.getMetadata());
          continue;
        } else {
          lastencoder = null;
        }
      }
      
      lastencoder = encoder;
    }

    if (null != lastencoder) {
      outputEncoder(lastencoder, chunksize, maxsize, suffix);      
    }
  }
  
  private static void outputEncoder(GTSEncoder encoder, long chunksize, long maxsize, String suffix) throws Exception {
    
    Metadata metadata = new Metadata(encoder.getMetadata());
    
    GTSDecoder decoder = encoder.getDecoder();
    GTSEncoder chunkEncoder = null;
    
    Long lastChunk = null;
    long maxTS = Long.MIN_VALUE;
    Long lastTs = null;
    Boolean order = null;
    
    while(decoder.next()) {
      long ts = decoder.getTimestamp();
      long chunk = ts >= 0 ? ts / chunksize : ((ts + 1) / chunksize) - 1;
      
      if (null == lastChunk) {
        lastChunk = chunk;
      }
      
      if (chunk != lastChunk) {
        outputChunk(chunkEncoder, maxTS, maxsize);
        chunkEncoder = null;
        maxTS = Long.MIN_VALUE;        
      }

      if (null == chunkEncoder) {
        chunkEncoder = new GTSEncoder(0L);
        chunkEncoder.setMetadata(metadata);
        if (null != suffix) {
          chunkEncoder.setName(chunkEncoder.getName() + suffix);
        }        
      }

      chunkEncoder.addValue(decoder.getTimestamp(), decoder.getLocation(), decoder.getElevation(), decoder.getValue());
      
      if (ts > maxTS) {
        maxTS = ts;
      }
      
      if (null == lastTs) {
        lastTs = ts;
      } else {
        if (null == order) {
          order = ts >= lastTs;
        } else if (order && ts < lastTs) {
          throw new RuntimeException("Timestamps moved unexpectedly backwards.");
        } else if (!order && ts > lastTs) {
          throw new RuntimeException("Timestamps moved unexpectedly forward.");
        }<commit_msg>Added missing carrying on of chunk<commit_after>package io.warp10.continuum;

import io.warp10.WarpConfig;
import io.warp10.continuum.gts.GTSDecoder;
import io.warp10.continuum.gts.GTSEncoder;
import io.warp10.continuum.gts.GTSHelper;
import io.warp10.continuum.gts.GTSWrapperHelper;
import io.warp10.continuum.store.Constants;
import io.warp10.continuum.store.thrift.data.GTSWrapper;
import io.warp10.continuum.store.thrift.data.Metadata;
import io.warp10.crypto.OrderPreservingBase64;

import java.io.BufferedReader;
import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map.Entry;
import java.util.Set;

import org.apache.commons.cli.CommandLine;
import org.apache.commons.cli.CommandLineParser;
import org.apache.commons.cli.DefaultParser;
import org.apache.commons.cli.Options;
import org.apache.thrift.TSerializer;
import org.apache.thrift.protocol.TCompactProtocol;
import org.python.jline.internal.InputStreamReader;

/**
 * Packs Geo Time Series into chunks spanning at most a given
 * timespan.
 */
public class Pack {
  
  private static final long MAX_ENCODER_SIZE = 16 * 1024 * 1024L;
  
  // Set of Geo Time Series already encountered
  private static Set<Metadata> parsed = new HashSet<Metadata>();

  private static void pack(long chunksize, int maxsize, String suffix) throws Exception {
    BufferedReader br = new BufferedReader(new InputStreamReader(System.in));
    
    long now = TimeSource.getTime();
        
    GTSEncoder lastencoder = null;
    
    while(true) {
      String line = br.readLine();
      
      if (null == line) {
        break;
      }
      
      GTSEncoder encoder = GTSHelper.parse(lastencoder, line, null, now);

      if (null != lastencoder && (encoder != lastencoder || lastencoder.size() > MAX_ENCODER_SIZE)) {
        // We changed the GTS, pack 'lastencoder'
        outputEncoder(lastencoder, chunksize, maxsize, suffix);
        
        if (encoder == lastencoder) {
          // We allocate a branch new Encoder but set its metadata to those of 'encoder'
          // so continuation lines are supported
          lastencoder = new GTSEncoder(0L);
          lastencoder.setMetadata(encoder.getMetadata());
          continue;
        } else {
          lastencoder = null;
        }
      }
      
      lastencoder = encoder;
    }

    if (null != lastencoder) {
      outputEncoder(lastencoder, chunksize, maxsize, suffix);      
    }
  }
  
  private static void outputEncoder(GTSEncoder encoder, long chunksize, long maxsize, String suffix) throws Exception {
    
    Metadata metadata = new Metadata(encoder.getMetadata());
    
    GTSDecoder decoder = encoder.getDecoder();
    GTSEncoder chunkEncoder = null;
    
    Long lastChunk = null;
    long maxTS = Long.MIN_VALUE;
    Long lastTs = null;
    Boolean order = null;
    
    while(decoder.next()) {
      long ts = decoder.getTimestamp();
      long chunk = ts >= 0 ? ts / chunksize : ((ts + 1) / chunksize) - 1;
      
      if (null == lastChunk) {
        lastChunk = chunk;
      }
      
      if (chunk != lastChunk) {
        outputChunk(chunkEncoder, maxTS, maxsize);
        chunkEncoder = null;
        lastChunk = chunk;
        maxTS = Long.MIN_VALUE;        
      }

      if (null == chunkEncoder) {
        chunkEncoder = new GTSEncoder(0L);
        chunkEncoder.setMetadata(metadata);
        if (null != suffix) {
          chunkEncoder.setName(chunkEncoder.getName() + suffix);
        }        
      }

      chunkEncoder.addValue(decoder.getTimestamp(), decoder.getLocation(), decoder.getElevation(), decoder.getValue());
      
      if (ts > maxTS) {
        maxTS = ts;
      }
      
      if (null == lastTs) {
        lastTs = ts;
      } else {
        if (null == order) {
          order = ts >= lastTs;
        } else if (order && ts < lastTs) {
          throw new RuntimeException("Timestamps moved unexpectedly backwards.");
        } else if (!order && ts > lastTs) {
          throw new RuntimeException("Timestamps moved unexpectedly forward.");
        }
{'fraction_non_alphanumeric': 0.10454323400097704, 'fraction_numerical': 0.00903761602344895, 'mean_word_length': 2.647661469933185, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 17, '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': '5930570', 'n_tokens_mistral': 2497, 'n_tokens_neox': 2399, 'n_words': 601}