if (!conn.tableOperations().exists(opts.getTableName())) {
throw new TableNotFoundException(null, opts.getTableName(), "Consult the README and create the table before starting ingest.");
}
BatchWriter bw = conn.createBatchWriter(opts.getTableName(), bwOpts.getBatchWriterConfig());
bw = Trace.wrapAll(bw, new CountSampler(1024));
Random r = new Random();
byte[] ingestInstanceId = UUID.randomUUID().toString().getBytes(Constants.UTF8);
System.out.printf("UUID %d %s%n", System.currentTimeMillis(), new String(ingestInstanceId, Constants.UTF8));
long count = 0;
final int flushInterval = 1000000;
final int maxDepth = 25;
// always want to point back to flushed data. This way the previous item should
// always exist in accumulo when verifying data. To do this make insert N point
// back to the row from insert (N - flushInterval). The array below is used to keep
// track of this.
long prevRows[] = new long[flushInterval];
long firstRows[] = new long[flushInterval];
int firstColFams[] = new int[flushInterval];
int firstColQuals[] = new int[flushInterval];
long lastFlushTime = System.currentTimeMillis();
out: while (true) {
// generate first set of nodes
ColumnVisibility cv = getVisibility(r);
for (int index = 0; index < flushInterval; index++) {
long rowLong = genLong(opts.min, opts.max, r);
prevRows[index] = rowLong;
firstRows[index] = rowLong;
int cf = r.nextInt(opts.maxColF);
int cq = r.nextInt(opts.maxColQ);
firstColFams[index] = cf;
firstColQuals[index] = cq;
Mutation m = genMutation(rowLong, cf, cq, cv, ingestInstanceId, count, null, r, opts.checksum);
count++;
bw.addMutation(m);
}
lastFlushTime = flush(bw, count, flushInterval, lastFlushTime);
if (count >= opts.num)
break out;
// generate subsequent sets of nodes that link to previous set of nodes
for (int depth = 1; depth < maxDepth; depth++) {
for (int index = 0; index < flushInterval; index++) {
long rowLong = genLong(opts.min, opts.max, r);
byte[] prevRow = genRow(prevRows[index]);
prevRows[index] = rowLong;
Mutation m = genMutation(rowLong, r.nextInt(opts.maxColF), r.nextInt(opts.maxColQ), cv, ingestInstanceId, count, prevRow, r, opts.checksum);
count++;
bw.addMutation(m);
}
lastFlushTime = flush(bw, count, flushInterval, lastFlushTime);
if (count >= opts.num)
break out;
}
// create one big linked list, this makes all of the first inserts
// point to something
for (int index = 0; index < flushInterval - 1; index++) {
Mutation m = genMutation(firstRows[index], firstColFams[index], firstColQuals[index], cv, ingestInstanceId, count, genRow(prevRows[index + 1]), r,
opts.checksum);
count++;
bw.addMutation(m);
}
lastFlushTime = flush(bw, count, flushInterval, lastFlushTime);
if (count >= opts.num)
break out;
}
bw.close();
opts.stopTracing();
}