@Override
public Predicate<Integer> project(Predicate<Long> predicate) {
  // year is the only time field that can be projected
  if (predicate instanceof Exists) {
    return Predicates.exists();
  } else if (predicate instanceof In) {
    return ((In<Long>) predicate).transform(this);
  } else if (predicate instanceof Range) {
    return Ranges.transformClosed(
        Ranges.adjustClosed(
            (Range<Long>) predicate, DiscreteDomains.longs()),
        this);
  } else {
    return null;
  }
}
 

@Override
public Predicate<Integer> project(Predicate<Integer> predicate) {
  if (predicate instanceof Exists) {
    return Predicates.exists();
  } else if (predicate instanceof In) {
    return ((In<Integer>) predicate).transform(this);
  } else if (predicate instanceof Range) {
    // must use a closed range:
    //   if this( 5 ) => 10 then this( 6 ) => 10, so 10 must be included
    return Ranges.transformClosed(
        Ranges.adjustClosed((Range<Integer>) predicate,
            DiscreteDomains.integers()), this);
  } else {
    return null;
  }
}
 

private TimeRangePredicateImpl(Range<Long> timeRange, boolean acceptEqual) {
  this.range = Ranges.adjustClosed(timeRange, DiscreteDomains.longs());
  this.acceptEqual = acceptEqual;

  int length = partitioners.size();
  this.names = new String[length];
  for (int i = 0; i < length; i += 1) {
    names[i] = partitioners.get(i).getName();
  }

  if (range.hasLowerBound()) {
    long start = range.lowerEndpoint() - (acceptEqual ? 0 : 1);
    this.lower = new int[length];
    for (int i = 0; i < length; i += 1) {
      lower[i] = partitioners.get(i).apply(start);
    }
  } else {
    this.lower = new int[0];
  }
  if (range.hasUpperBound()) {
    long stop = range.upperEndpoint() + (acceptEqual ? 0 : 1);
    this.upper = new int[length];
    for (int i = 0; i < length; i += 1) {
      upper[i] = partitioners.get(i).apply(stop);
    }
  } else {
    this.upper = new int[0];
  }
}
 

@Override
public Predicate<Long> project(Predicate<Long> predicate) {
  if (predicate instanceof Exists) {
    return Predicates.exists();
  } else if (predicate instanceof In) {
    return ((In<Long>) predicate).transform(this);
  } else if (predicate instanceof Range) {
    return Ranges.transformClosed(
        Ranges.adjustClosed((Range<Long>) predicate,
            DiscreteDomains.longs()), this);
  } else {
    return null;
  }
}
 

/**
 * Helper method to restart instances of runnables.
 */
private void restartRunnableInstances(final String runnableName, @Nullable final Set<Integer> instanceIds,
                                      final Runnable completion) {
  instanceChangeExecutor.execute(new Runnable() {
    @Override
    public void run() {
      LOG.debug("Begin restart runnable {} instances.", runnableName);
      int runningCount = runningContainers.count(runnableName);
      Set<Integer> instancesToRemove = instanceIds == null ? null : ImmutableSet.copyOf(instanceIds);
      if (instancesToRemove == null) {
        instancesToRemove = Ranges.closedOpen(0, runningCount).asSet(DiscreteDomains.integers());
      }

      LOG.info("Restarting instances {} for runnable {}", instancesToRemove, runnableName);
      RunnableContainerRequest containerRequest =
        createRunnableContainerRequest(runnableName, instancesToRemove.size(), false);
      runnableContainerRequests.add(containerRequest);

      for (int instanceId : instancesToRemove) {
        LOG.debug("Stop instance {} for runnable {}", instanceId, runnableName);
        try {
          runningContainers.stopByIdAndWait(runnableName, instanceId);
        } catch (Exception ex) {
          // could be thrown if the container already stopped.
          LOG.info("Exception thrown when stopping instance {} probably already stopped.", instanceId);
        }
      }

      LOG.info("All instances in {} for runnable {} are stopped. Ready to provision",
               instancesToRemove, runnableName);

      // set the container request to be ready
      containerRequest.setReadyToBeProvisioned();

      // For all runnables that needs to re-request for containers, update the expected count timestamp
      // so that the EventHandler would be triggered with the right expiration timestamp.
      expectedContainers.updateRequestTime(Collections.singleton(runnableName));

      completion.run();
    }
  });
}
 

@Override
public int run(String[] args) throws Exception {
  // going to generate a lot of random log messages
  final Random rand = new Random();

  // data is written to the staging dataset
  Dataset<Record> staging = Datasets.load(
      "dataset:file:/tmp/data/logs_staging", Record.class);

  // this is going to build our simple log records
  GenericRecordBuilder builder = new GenericRecordBuilder(
      staging.getDescriptor().getSchema());

  // generate timestamps 1 second apart starting 1 day ago
  final Calendar now = Calendar.getInstance(TimeZone.getTimeZone("UTC"));
  final long yesterday = now.getTimeInMillis() - DAY_IN_MILLIS;

  DatasetWriter<Record> writer = null;
  try {
    writer = staging.newWriter();

    // generate 15,000 messages, each 5 seconds apart, starting 24 hours ago
    // this is a little less than 24 hours worth of messages
    for (int second : Ranges.closed(0, 15000).asSet(DiscreteDomains.integers())) {
      LOG.info("Generating log message " + second);

      builder.set("timestamp", yesterday + second * 5000);
      builder.set("component", "GenerateSimpleLogs");

      int level = rand.nextInt(LOG_LEVELS.length);
      builder.set("level", LOG_LEVELS[level]);
      builder.set("message", LOG_MESSAGES[level]);

      writer.write(builder.build());
    }

    if (writer instanceof Flushable) {
      ((Flushable) writer).flush();
    }
  } finally {
    if (writer != null) {
      writer.close();
    }
  }

  return 0;
}
 

@Override
public String maxValue() {
  DiscreteDomains.integers();
  return upperBounds.get(upperBounds.size() - 1);
}
 

@Override
public Predicate<Long> projectStrict(Predicate<Long> predicate) {
  if (predicate instanceof Exists) {
    return Predicates.exists();
  } else if (predicate instanceof In) {
    Set<Long> possibleValues = Sets.newHashSet();
    In<Long> in = ((In<Long>) predicate).transform(this);
    for (Long val : Predicates.asSet(in)) {
      boolean matchedAll = true;
      for (long i = 0; i < size; i++) {
        matchedAll = matchedAll && predicate.apply(val + i);
      }
      if (matchedAll) {
        possibleValues.add(val);
      }
    }
    if (!possibleValues.isEmpty()) {
      return Predicates.in(possibleValues);
    }
  } else if (predicate instanceof Range) {
    Range<Long> closed = Ranges.adjustClosed(
        (Range<Long>) predicate, DiscreteDomains.longs());
    Long start = null;
    if (closed.hasLowerBound()) {
      if ((closed.lowerEndpoint() % size) == 0) {
        // the entire set of values is included
        start = closed.lowerEndpoint();
      } else {
        // start the predicate at the next value
        start = apply(closed.lowerEndpoint() + size);
      }
    }
    Long end = null;
    if (closed.hasUpperBound()) {
      if (((closed.upperEndpoint() + 1) % size) == 0) {
        // all values are included
        end = apply(closed.upperEndpoint());
      } else {
        // end the predicate at the previous value
        end = apply(closed.upperEndpoint() - size);
      }
    }
    if (start != null && end != null && start > end) {
      return null;
    }
    return Ranges.closed(start, end); // null start or end => unbound
  }
  return null;
}