apply new code formatter and save action
This commit is contained in:
@@ -6,7 +6,7 @@ import java.util.concurrent.TimeoutException;
|
||||
|
||||
public interface BlockingIterator<E> {
|
||||
|
||||
public Optional<E> next(long timeout, TimeUnit unit) throws InterruptedException, TimeoutException;
|
||||
public Optional<E> next(long timeout, TimeUnit unit) throws InterruptedException, TimeoutException;
|
||||
|
||||
public Optional<E> next() throws InterruptedException;
|
||||
public Optional<E> next() throws InterruptedException;
|
||||
}
|
||||
@@ -6,24 +6,24 @@ import java.util.concurrent.TimeUnit;
|
||||
|
||||
final class BlockingIteratorIterator<E> implements BlockingIterator<E> {
|
||||
|
||||
private final Iterator<E> iterator;
|
||||
private final Iterator<E> iterator;
|
||||
|
||||
public BlockingIteratorIterator(final Iterator<E> iterator) {
|
||||
this.iterator = iterator;
|
||||
}
|
||||
public BlockingIteratorIterator(final Iterator<E> iterator) {
|
||||
this.iterator = iterator;
|
||||
}
|
||||
|
||||
@Override
|
||||
public Optional<E> next() throws InterruptedException {
|
||||
if (iterator.hasNext()) {
|
||||
final E next = iterator.next();
|
||||
return Optional.of(next);
|
||||
} else {
|
||||
return Optional.empty();
|
||||
}
|
||||
}
|
||||
@Override
|
||||
public Optional<E> next() throws InterruptedException {
|
||||
if (iterator.hasNext()) {
|
||||
final E next = iterator.next();
|
||||
return Optional.of(next);
|
||||
} else {
|
||||
return Optional.empty();
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public Optional<E> next(final long timeout, final TimeUnit unit) throws InterruptedException {
|
||||
return next();
|
||||
}
|
||||
@Override
|
||||
public Optional<E> next(final long timeout, final TimeUnit unit) throws InterruptedException {
|
||||
return next();
|
||||
}
|
||||
}
|
||||
@@ -10,50 +10,50 @@ import org.slf4j.LoggerFactory;
|
||||
|
||||
public final class BlockingQueueIterator<E> implements BlockingIterator<E> {
|
||||
|
||||
private static final Logger LOGGER = LoggerFactory.getLogger(BlockingQueueIterator.class);
|
||||
private static final Logger LOGGER = LoggerFactory.getLogger(BlockingQueueIterator.class);
|
||||
|
||||
private final BlockingQueue<E> queue;
|
||||
private final BlockingQueue<E> queue;
|
||||
|
||||
private boolean closed = false;
|
||||
private boolean closed = false;
|
||||
|
||||
private final E poison;
|
||||
private final E poison;
|
||||
|
||||
public BlockingQueueIterator(final BlockingQueue<E> queue, final E poison) {
|
||||
this.queue = queue;
|
||||
this.poison = poison;
|
||||
}
|
||||
public BlockingQueueIterator(final BlockingQueue<E> queue, final E poison) {
|
||||
this.queue = queue;
|
||||
this.poison = poison;
|
||||
}
|
||||
|
||||
@Override
|
||||
public Optional<E> next() throws InterruptedException {
|
||||
try {
|
||||
return next(Long.MAX_VALUE, TimeUnit.NANOSECONDS);
|
||||
} catch (final TimeoutException e) {
|
||||
throw new IllegalStateException(
|
||||
"We just got a timeout exception after waiting the longest time possible. Which is "
|
||||
+ TimeUnit.NANOSECONDS.toDays(Long.MAX_VALUE) + " days. We didn't expect that.",
|
||||
e);
|
||||
}
|
||||
}
|
||||
@Override
|
||||
public Optional<E> next() throws InterruptedException {
|
||||
try {
|
||||
return next(Long.MAX_VALUE, TimeUnit.NANOSECONDS);
|
||||
} catch (final TimeoutException e) {
|
||||
throw new IllegalStateException(
|
||||
"We just got a timeout exception after waiting the longest time possible. Which is "
|
||||
+ TimeUnit.NANOSECONDS.toDays(Long.MAX_VALUE) + " days. We didn't expect that.",
|
||||
e);
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public Optional<E> next(final long timeout, final TimeUnit unit) throws InterruptedException, TimeoutException {
|
||||
@Override
|
||||
public Optional<E> next(final long timeout, final TimeUnit unit) throws InterruptedException, TimeoutException {
|
||||
|
||||
if (closed) {
|
||||
return Optional.empty();
|
||||
}
|
||||
if (closed) {
|
||||
return Optional.empty();
|
||||
}
|
||||
|
||||
LOGGER.trace("wait for next entry");
|
||||
final E next = queue.poll(timeout, unit);
|
||||
LOGGER.trace("received entry: {}", next);
|
||||
LOGGER.trace("wait for next entry");
|
||||
final E next = queue.poll(timeout, unit);
|
||||
LOGGER.trace("received entry: {}", next);
|
||||
|
||||
if (next == poison) {
|
||||
LOGGER.trace("received poison");
|
||||
closed = true;
|
||||
return Optional.empty();
|
||||
} else if (next == null) {
|
||||
throw new TimeoutException();
|
||||
}
|
||||
if (next == poison) {
|
||||
LOGGER.trace("received poison");
|
||||
closed = true;
|
||||
return Optional.empty();
|
||||
} else if (next == null) {
|
||||
throw new TimeoutException();
|
||||
}
|
||||
|
||||
return Optional.of(next);
|
||||
}
|
||||
return Optional.of(next);
|
||||
}
|
||||
}
|
||||
@@ -7,20 +7,20 @@ import org.lucares.pdb.api.Entry;
|
||||
|
||||
public class EntryToEntriesIterator implements Iterator<Entries> {
|
||||
|
||||
private final Iterator<Entry> entryIterator;
|
||||
private final Iterator<Entry> entryIterator;
|
||||
|
||||
public EntryToEntriesIterator(final Iterator<Entry> entryIterator) {
|
||||
this.entryIterator = entryIterator;
|
||||
}
|
||||
public EntryToEntriesIterator(final Iterator<Entry> entryIterator) {
|
||||
this.entryIterator = entryIterator;
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean hasNext() {
|
||||
return entryIterator.hasNext();
|
||||
}
|
||||
@Override
|
||||
public boolean hasNext() {
|
||||
return entryIterator.hasNext();
|
||||
}
|
||||
|
||||
@Override
|
||||
public Entries next() {
|
||||
return new Entries(entryIterator.next());
|
||||
}
|
||||
@Override
|
||||
public Entries next() {
|
||||
return new Entries(entryIterator.next());
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -6,30 +6,30 @@ import org.lucares.pdb.api.Tags;
|
||||
import org.lucares.pdb.datastore.PdbFile;
|
||||
|
||||
class Group {
|
||||
private final Tags tags;
|
||||
private final Tags tags;
|
||||
|
||||
private final List<PdbFile> files;
|
||||
private final List<PdbFile> files;
|
||||
|
||||
public Group(final Tags tags, final List<PdbFile> files) {
|
||||
super();
|
||||
this.tags = tags;
|
||||
this.files = files;
|
||||
}
|
||||
public Group(final Tags tags, final List<PdbFile> files) {
|
||||
super();
|
||||
this.tags = tags;
|
||||
this.files = files;
|
||||
}
|
||||
|
||||
public Tags getTags() {
|
||||
return tags;
|
||||
}
|
||||
public Tags getTags() {
|
||||
return tags;
|
||||
}
|
||||
|
||||
public List<PdbFile> getFiles() {
|
||||
return files;
|
||||
}
|
||||
public List<PdbFile> getFiles() {
|
||||
return files;
|
||||
}
|
||||
|
||||
public void addFile(final PdbFile file) {
|
||||
files.add(file);
|
||||
}
|
||||
public void addFile(final PdbFile file) {
|
||||
files.add(file);
|
||||
}
|
||||
|
||||
@Override
|
||||
public String toString() {
|
||||
return tags + ": " + files.size() + " files";
|
||||
}
|
||||
@Override
|
||||
public String toString() {
|
||||
return tags + ": " + files.size() + " files";
|
||||
}
|
||||
}
|
||||
|
||||
@@ -13,61 +13,61 @@ import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
||||
public class Grouping {
|
||||
private static final Logger LOGGER = LoggerFactory.getLogger(Grouping.class);
|
||||
private static final Logger LOGGER = LoggerFactory.getLogger(Grouping.class);
|
||||
|
||||
public static final List<String> NO_GROUPING = Collections.emptyList();
|
||||
public static final List<String> NO_GROUPING = Collections.emptyList();
|
||||
|
||||
private final List<Group> groups = new ArrayList<>();
|
||||
private final List<Group> groups = new ArrayList<>();
|
||||
|
||||
private Grouping(final Group group) {
|
||||
this.groups.add(group);
|
||||
}
|
||||
private Grouping(final Group group) {
|
||||
this.groups.add(group);
|
||||
}
|
||||
|
||||
private Grouping(final Collection<Group> groups) {
|
||||
this.groups.addAll(groups);
|
||||
}
|
||||
private Grouping(final Collection<Group> groups) {
|
||||
this.groups.addAll(groups);
|
||||
}
|
||||
|
||||
public static Grouping groupBy(final List<PdbFile> pdbFiles, final List<String> groupByField) {
|
||||
public static Grouping groupBy(final List<PdbFile> pdbFiles, final List<String> groupByField) {
|
||||
|
||||
final Grouping result;
|
||||
if (noGrouping(groupByField)) {
|
||||
final Group group = new Group(Tags.EMPTY, pdbFiles);
|
||||
final Grouping result;
|
||||
if (noGrouping(groupByField)) {
|
||||
final Group group = new Group(Tags.EMPTY, pdbFiles);
|
||||
|
||||
result = new Grouping(group);
|
||||
} else {
|
||||
final Map<Tags, Group> grouping = new HashMap<>();
|
||||
result = new Grouping(group);
|
||||
} else {
|
||||
final Map<Tags, Group> grouping = new HashMap<>();
|
||||
|
||||
for (final PdbFile pdbFile : pdbFiles) {
|
||||
final Tags tags = pdbFile.getTags();
|
||||
final Tags groupTags = tags.subset(groupByField);
|
||||
for (final PdbFile pdbFile : pdbFiles) {
|
||||
final Tags tags = pdbFile.getTags();
|
||||
final Tags groupTags = tags.subset(groupByField);
|
||||
|
||||
addIfNotExists(grouping, groupTags);
|
||||
grouping.get(groupTags).addFile(pdbFile);
|
||||
}
|
||||
result = new Grouping(grouping.values());
|
||||
}
|
||||
LOGGER.trace("grouped {} files by {}: {}", pdbFiles.size(), groupByField, result);
|
||||
return result;
|
||||
}
|
||||
addIfNotExists(grouping, groupTags);
|
||||
grouping.get(groupTags).addFile(pdbFile);
|
||||
}
|
||||
result = new Grouping(grouping.values());
|
||||
}
|
||||
LOGGER.trace("grouped {} files by {}: {}", pdbFiles.size(), groupByField, result);
|
||||
return result;
|
||||
}
|
||||
|
||||
private static boolean noGrouping(final List<String> groupByField) {
|
||||
return groupByField == null || groupByField.isEmpty();
|
||||
}
|
||||
private static boolean noGrouping(final List<String> groupByField) {
|
||||
return groupByField == null || groupByField.isEmpty();
|
||||
}
|
||||
|
||||
private static void addIfNotExists(final Map<Tags, Group> grouping, final Tags groupTags) {
|
||||
if (!grouping.containsKey(groupTags)) {
|
||||
final List<PdbFile> files = new ArrayList<>();
|
||||
private static void addIfNotExists(final Map<Tags, Group> grouping, final Tags groupTags) {
|
||||
if (!grouping.containsKey(groupTags)) {
|
||||
final List<PdbFile> files = new ArrayList<>();
|
||||
|
||||
grouping.put(groupTags, new Group(groupTags, files));
|
||||
}
|
||||
}
|
||||
grouping.put(groupTags, new Group(groupTags, files));
|
||||
}
|
||||
}
|
||||
|
||||
public Collection<Group> getGroups() {
|
||||
return groups;
|
||||
}
|
||||
public Collection<Group> getGroups() {
|
||||
return groups;
|
||||
}
|
||||
|
||||
@Override
|
||||
public String toString() {
|
||||
return String.valueOf(groups);
|
||||
}
|
||||
@Override
|
||||
public String toString() {
|
||||
return String.valueOf(groups);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -31,153 +31,153 @@ import org.slf4j.LoggerFactory;
|
||||
|
||||
public class PdbExport {
|
||||
|
||||
private static final int KB = 1024;
|
||||
private static final int MB = KB * 1024;
|
||||
private static final int GB = MB * 1024;
|
||||
private static final int KB = 1024;
|
||||
private static final int MB = KB * 1024;
|
||||
private static final int GB = MB * 1024;
|
||||
|
||||
public static final char MAGIC_BYTE = '#';
|
||||
public static final char MARKER_DICT_ENTRY_CHAR = '$';
|
||||
public static final String MARKER_DICT_ENTRY = String.valueOf(MARKER_DICT_ENTRY_CHAR);
|
||||
public static final char SEPARATOR_TAG_ID_CHAR = ':';
|
||||
public static final String SEPARATOR_TAG_ID = String.valueOf(SEPARATOR_TAG_ID_CHAR);
|
||||
public static final char MAGIC_BYTE = '#';
|
||||
public static final char MARKER_DICT_ENTRY_CHAR = '$';
|
||||
public static final String MARKER_DICT_ENTRY = String.valueOf(MARKER_DICT_ENTRY_CHAR);
|
||||
public static final char SEPARATOR_TAG_ID_CHAR = ':';
|
||||
public static final String SEPARATOR_TAG_ID = String.valueOf(SEPARATOR_TAG_ID_CHAR);
|
||||
|
||||
private static final Logger LOGGER = LoggerFactory.getLogger(PdbExport.class);
|
||||
private static final Logger LOGGER = LoggerFactory.getLogger(PdbExport.class);
|
||||
|
||||
public static void main(final String[] args) throws Exception {
|
||||
public static void main(final String[] args) throws Exception {
|
||||
|
||||
initLogging();
|
||||
initLogging();
|
||||
|
||||
final Path dataDirectory = Paths.get(args[0]);
|
||||
final Path backupDir = Paths.get(args[1])
|
||||
.resolve(OffsetDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd_HH-mm-ss")));
|
||||
final Path dataDirectory = Paths.get(args[0]);
|
||||
final Path backupDir = Paths.get(args[1])
|
||||
.resolve(OffsetDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd_HH-mm-ss")));
|
||||
|
||||
export(dataDirectory, backupDir);
|
||||
}
|
||||
export(dataDirectory, backupDir);
|
||||
}
|
||||
|
||||
public static List<Path> export(final Path dataDirectory, final Path backupDir) throws Exception {
|
||||
final List<Path> exportFiles = new ArrayList<>();
|
||||
Files.createDirectories(backupDir);
|
||||
public static List<Path> export(final Path dataDirectory, final Path backupDir) throws Exception {
|
||||
final List<Path> exportFiles = new ArrayList<>();
|
||||
Files.createDirectories(backupDir);
|
||||
|
||||
Runtime.getRuntime().addShutdownHook(new Thread() {
|
||||
@Override
|
||||
public void run() {
|
||||
LOGGER.info("shutdown hook");
|
||||
}
|
||||
Runtime.getRuntime().addShutdownHook(new Thread() {
|
||||
@Override
|
||||
public void run() {
|
||||
LOGGER.info("shutdown hook");
|
||||
}
|
||||
|
||||
});
|
||||
});
|
||||
|
||||
final OffsetDateTime start = OffsetDateTime.now();
|
||||
final String datePrefix = start.format(DateTimeFormatter.ofPattern("yyyy-MM-dd_HH-mm-ss"));
|
||||
final AtomicLong tagsIdCounter = new AtomicLong(0);
|
||||
long exportFileCounter = 0;
|
||||
final OffsetDateTime start = OffsetDateTime.now();
|
||||
final String datePrefix = start.format(DateTimeFormatter.ofPattern("yyyy-MM-dd_HH-mm-ss"));
|
||||
final AtomicLong tagsIdCounter = new AtomicLong(0);
|
||||
long exportFileCounter = 0;
|
||||
|
||||
Path exportFile = null;
|
||||
Writer writer = null;
|
||||
Path exportFile = null;
|
||||
Writer writer = null;
|
||||
|
||||
try (final PerformanceDb db = new PerformanceDb(dataDirectory);) {
|
||||
try (final PerformanceDb db = new PerformanceDb(dataDirectory);) {
|
||||
|
||||
LOGGER.info("Searching for all files. This may take a while ...");
|
||||
final List<PdbFile> pdbFiles = db.getFilesForQuery(new Query("", DateTimeRange.max()));
|
||||
LOGGER.info("Searching for all files. This may take a while ...");
|
||||
final List<PdbFile> pdbFiles = db.getFilesForQuery(new Query("", DateTimeRange.max()));
|
||||
|
||||
long count = 0;
|
||||
long lastEpochMilli = 0;
|
||||
long begin = System.currentTimeMillis();
|
||||
long count = 0;
|
||||
long lastEpochMilli = 0;
|
||||
long begin = System.currentTimeMillis();
|
||||
|
||||
for (final PdbFile pdbFile : pdbFiles) {
|
||||
for (final PdbFile pdbFile : pdbFiles) {
|
||||
|
||||
if (writer == null || Files.size(exportFile) > 4 * GB) {
|
||||
if (writer != null) {
|
||||
writer.flush();
|
||||
writer.close();
|
||||
}
|
||||
exportFile = backupDir.resolve(String.format("%s.%05d.pdb.gz", datePrefix, exportFileCounter++));
|
||||
exportFiles.add(exportFile);
|
||||
writer = createWriter(exportFile);
|
||||
LOGGER.info("new export file: {}", exportFile);
|
||||
if (writer == null || Files.size(exportFile) > 4 * GB) {
|
||||
if (writer != null) {
|
||||
writer.flush();
|
||||
writer.close();
|
||||
}
|
||||
exportFile = backupDir.resolve(String.format("%s.%05d.pdb.gz", datePrefix, exportFileCounter++));
|
||||
exportFiles.add(exportFile);
|
||||
writer = createWriter(exportFile);
|
||||
LOGGER.info("new export file: {}", exportFile);
|
||||
|
||||
lastEpochMilli = 0;
|
||||
}
|
||||
lastEpochMilli = 0;
|
||||
}
|
||||
|
||||
final Stream<LongList> timeValueStream = PdbFile.toStream(Arrays.asList(pdbFile), db.getDataStore());
|
||||
final Stream<LongList> timeValueStream = PdbFile.toStream(Arrays.asList(pdbFile), db.getDataStore());
|
||||
|
||||
final Tags tags = pdbFile.getTags();
|
||||
final long tagsId = addNewTagsToDictionary(writer, tags, tagsIdCounter);
|
||||
final Tags tags = pdbFile.getTags();
|
||||
final long tagsId = addNewTagsToDictionary(writer, tags, tagsIdCounter);
|
||||
|
||||
final Iterator<LongList> it = timeValueStream.iterator();
|
||||
while (it.hasNext()) {
|
||||
final LongList entry = it.next();
|
||||
final Iterator<LongList> it = timeValueStream.iterator();
|
||||
while (it.hasNext()) {
|
||||
final LongList entry = it.next();
|
||||
|
||||
for (int i = 0; i < entry.size(); i += 2) {
|
||||
for (int i = 0; i < entry.size(); i += 2) {
|
||||
|
||||
final long epochMilli = entry.get(i);
|
||||
final long value = entry.get(i + 1);
|
||||
final long epochMilli = entry.get(i);
|
||||
final long value = entry.get(i + 1);
|
||||
|
||||
final long epochMilliDiff = epochMilli - lastEpochMilli;
|
||||
lastEpochMilli = epochMilli;
|
||||
final long epochMilliDiff = epochMilli - lastEpochMilli;
|
||||
lastEpochMilli = epochMilli;
|
||||
|
||||
writer.write(Long.toString(epochMilliDiff));
|
||||
writer.write(',');
|
||||
writer.write(Long.toString(value));
|
||||
writer.write(',');
|
||||
writer.write(Long.toString(tagsId));
|
||||
writer.write('\n');
|
||||
writer.write(Long.toString(epochMilliDiff));
|
||||
writer.write(',');
|
||||
writer.write(Long.toString(value));
|
||||
writer.write(',');
|
||||
writer.write(Long.toString(tagsId));
|
||||
writer.write('\n');
|
||||
|
||||
count++;
|
||||
final long chunk = 10_000_000;
|
||||
if (count % chunk == 0) {
|
||||
final long end = System.currentTimeMillis();
|
||||
final long duration = end - begin;
|
||||
final long entriesPerSecond = (long) (chunk / (duration / 1000.0));
|
||||
LOGGER.info("progress: {} - {} entries/s + duration {}", String.format("%,d", count),
|
||||
String.format("%,d", entriesPerSecond), duration);
|
||||
begin = System.currentTimeMillis();
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
count++;
|
||||
final long chunk = 10_000_000;
|
||||
if (count % chunk == 0) {
|
||||
final long end = System.currentTimeMillis();
|
||||
final long duration = end - begin;
|
||||
final long entriesPerSecond = (long) (chunk / (duration / 1000.0));
|
||||
LOGGER.info("progress: {} - {} entries/s + duration {}", String.format("%,d", count),
|
||||
String.format("%,d", entriesPerSecond), duration);
|
||||
begin = System.currentTimeMillis();
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
LOGGER.info("total: " + count);
|
||||
LOGGER.info("total: " + count);
|
||||
|
||||
} finally {
|
||||
if (writer != null) {
|
||||
writer.close();
|
||||
}
|
||||
}
|
||||
} finally {
|
||||
if (writer != null) {
|
||||
writer.close();
|
||||
}
|
||||
}
|
||||
|
||||
final OffsetDateTime end = OffsetDateTime.now();
|
||||
final OffsetDateTime end = OffsetDateTime.now();
|
||||
|
||||
LOGGER.info("duration: " + Duration.between(start, end));
|
||||
return exportFiles;
|
||||
}
|
||||
LOGGER.info("duration: " + Duration.between(start, end));
|
||||
return exportFiles;
|
||||
}
|
||||
|
||||
private static void initLogging() {
|
||||
Configurator.setRootLevel(Level.INFO);
|
||||
}
|
||||
private static void initLogging() {
|
||||
Configurator.setRootLevel(Level.INFO);
|
||||
}
|
||||
|
||||
private static long addNewTagsToDictionary(final Writer writer, final Tags tags, final AtomicLong tagsIdCounter)
|
||||
throws IOException {
|
||||
final long tagsId = tagsIdCounter.getAndIncrement();
|
||||
private static long addNewTagsToDictionary(final Writer writer, final Tags tags, final AtomicLong tagsIdCounter)
|
||||
throws IOException {
|
||||
final long tagsId = tagsIdCounter.getAndIncrement();
|
||||
|
||||
writer.write(MARKER_DICT_ENTRY);
|
||||
writer.write(Long.toString(tagsId));
|
||||
writer.write(SEPARATOR_TAG_ID);
|
||||
writer.write(tags.toCsv());
|
||||
writer.write('\n');
|
||||
writer.write(MARKER_DICT_ENTRY);
|
||||
writer.write(Long.toString(tagsId));
|
||||
writer.write(SEPARATOR_TAG_ID);
|
||||
writer.write(tags.toCsv());
|
||||
writer.write('\n');
|
||||
|
||||
return tagsId;
|
||||
}
|
||||
return tagsId;
|
||||
}
|
||||
|
||||
private static Writer createWriter(final Path file) {
|
||||
private static Writer createWriter(final Path file) {
|
||||
|
||||
try {
|
||||
final OutputStreamWriter writer = new OutputStreamWriter(
|
||||
new GZIPOutputStream(new FileOutputStream(file.toFile()), 4096 * 4), StandardCharsets.UTF_8);
|
||||
// initialize file header
|
||||
writer.write(MAGIC_BYTE);
|
||||
return writer;
|
||||
try {
|
||||
final OutputStreamWriter writer = new OutputStreamWriter(
|
||||
new GZIPOutputStream(new FileOutputStream(file.toFile()), 4096 * 4), StandardCharsets.UTF_8);
|
||||
// initialize file header
|
||||
writer.write(MAGIC_BYTE);
|
||||
return writer;
|
||||
|
||||
} catch (final IOException e) {
|
||||
throw new IllegalStateException(e);
|
||||
}
|
||||
}
|
||||
} catch (final IOException e) {
|
||||
throw new IllegalStateException(e);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -30,157 +30,157 @@ import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
||||
public class PerformanceDb implements AutoCloseable {
|
||||
private final static Logger LOGGER = LoggerFactory.getLogger(PerformanceDb.class);
|
||||
private final static Logger METRICS_LOGGER = LoggerFactory.getLogger("org.lucares.metrics.ingestion.block");
|
||||
private final static Logger LOGGER = LoggerFactory.getLogger(PerformanceDb.class);
|
||||
private final static Logger METRICS_LOGGER = LoggerFactory.getLogger("org.lucares.metrics.ingestion.block");
|
||||
|
||||
private final DataStore dataStore;
|
||||
private final DataStore dataStore;
|
||||
|
||||
public PerformanceDb(final Path dataDirectory) throws IOException {
|
||||
public PerformanceDb(final Path dataDirectory) throws IOException {
|
||||
|
||||
dataStore = new DataStore(dataDirectory);
|
||||
dataStore = new DataStore(dataDirectory);
|
||||
|
||||
}
|
||||
}
|
||||
|
||||
void putEntry(final Entry entry) throws WriteException {
|
||||
putEntries(Arrays.asList(entry));
|
||||
}
|
||||
void putEntry(final Entry entry) throws WriteException {
|
||||
putEntries(Arrays.asList(entry));
|
||||
}
|
||||
|
||||
void putEntries(final Iterable<Entry> entries) throws WriteException {
|
||||
putEntries(entries.iterator());
|
||||
}
|
||||
void putEntries(final Iterable<Entry> entries) throws WriteException {
|
||||
putEntries(entries.iterator());
|
||||
}
|
||||
|
||||
private void putEntries(final Iterator<Entry> entries) throws WriteException {
|
||||
private void putEntries(final Iterator<Entry> entries) throws WriteException {
|
||||
|
||||
final EntryToEntriesIterator entriesIterator = new EntryToEntriesIterator(entries);
|
||||
final BlockingIteratorIterator<Entries> iterator = new BlockingIteratorIterator<>(entriesIterator);
|
||||
putEntries(iterator);
|
||||
}
|
||||
final EntryToEntriesIterator entriesIterator = new EntryToEntriesIterator(entries);
|
||||
final BlockingIteratorIterator<Entries> iterator = new BlockingIteratorIterator<>(entriesIterator);
|
||||
putEntries(iterator);
|
||||
}
|
||||
|
||||
public void putEntries(final BlockingIterator<Entries> entriesIterator) throws WriteException {
|
||||
public void putEntries(final BlockingIterator<Entries> entriesIterator) throws WriteException {
|
||||
|
||||
final Duration timeBetweenSyncs = Duration.ofSeconds(1);
|
||||
long count = 0;
|
||||
long insertionsSinceLastSync = 0;
|
||||
final Duration timeBetweenSyncs = Duration.ofSeconds(1);
|
||||
long count = 0;
|
||||
long insertionsSinceLastSync = 0;
|
||||
|
||||
try {
|
||||
long lastSync = System.currentTimeMillis();
|
||||
long nextSync = lastSync + timeBetweenSyncs.toMillis();
|
||||
try {
|
||||
long lastSync = System.currentTimeMillis();
|
||||
long nextSync = lastSync + timeBetweenSyncs.toMillis();
|
||||
|
||||
while (true) {
|
||||
final Optional<Entries> entriesOptional = entriesIterator.next();
|
||||
if (!entriesOptional.isPresent()) {
|
||||
break;
|
||||
}
|
||||
while (true) {
|
||||
final Optional<Entries> entriesOptional = entriesIterator.next();
|
||||
if (!entriesOptional.isPresent()) {
|
||||
break;
|
||||
}
|
||||
|
||||
final Entries entries = entriesOptional.get();
|
||||
for (final Entry entry : entries) {
|
||||
final Entries entries = entriesOptional.get();
|
||||
for (final Entry entry : entries) {
|
||||
|
||||
try {
|
||||
final Tags tags = entry.getTags();
|
||||
final long dateAsEpochMilli = entry.getEpochMilli();
|
||||
final long value = entry.getValue();
|
||||
try {
|
||||
final Tags tags = entry.getTags();
|
||||
final long dateAsEpochMilli = entry.getEpochMilli();
|
||||
final long value = entry.getValue();
|
||||
|
||||
dataStore.write(dateAsEpochMilli, tags, value);
|
||||
dataStore.write(dateAsEpochMilli, tags, value);
|
||||
|
||||
count++;
|
||||
insertionsSinceLastSync++;
|
||||
count++;
|
||||
insertionsSinceLastSync++;
|
||||
|
||||
if (nextSync <= System.currentTimeMillis()) {
|
||||
final long end = System.currentTimeMillis();
|
||||
final long duration = end - lastSync;
|
||||
final long entriesPerSecond = (long) (insertionsSinceLastSync / (duration / 1000.0));
|
||||
if (nextSync <= System.currentTimeMillis()) {
|
||||
final long end = System.currentTimeMillis();
|
||||
final long duration = end - lastSync;
|
||||
final long entriesPerSecond = (long) (insertionsSinceLastSync / (duration / 1000.0));
|
||||
|
||||
METRICS_LOGGER.debug(String.format("inserting %d/s ; total: %,d; last: %s",
|
||||
entriesPerSecond, count, entry));
|
||||
METRICS_LOGGER.debug(String.format("inserting %d/s ; total: %,d; last: %s",
|
||||
entriesPerSecond, count, entry));
|
||||
|
||||
lastSync = System.currentTimeMillis();
|
||||
nextSync = lastSync + timeBetweenSyncs.toMillis();
|
||||
insertionsSinceLastSync = 0;
|
||||
}
|
||||
lastSync = System.currentTimeMillis();
|
||||
nextSync = lastSync + timeBetweenSyncs.toMillis();
|
||||
insertionsSinceLastSync = 0;
|
||||
}
|
||||
|
||||
} catch (final InvalidValueException | SyntaxException e) {
|
||||
} catch (final InvalidValueException | SyntaxException e) {
|
||||
|
||||
LOGGER.info("skipping entry: " + e.getMessage() + " : " + entry);
|
||||
LOGGER.info("", e);
|
||||
}
|
||||
}
|
||||
}
|
||||
LOGGER.info("skipping entry: " + e.getMessage() + " : " + entry);
|
||||
LOGGER.info("", e);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
} catch (final RuntimeException e) {
|
||||
throw new WriteException(e);
|
||||
} catch (final InterruptedException e) {
|
||||
Thread.currentThread().interrupt();
|
||||
LOGGER.info("Thread was interrupted. Aborting exectution.");
|
||||
} finally {
|
||||
dataStore.flush();
|
||||
}
|
||||
}
|
||||
} catch (final RuntimeException e) {
|
||||
throw new WriteException(e);
|
||||
} catch (final InterruptedException e) {
|
||||
Thread.currentThread().interrupt();
|
||||
LOGGER.info("Thread was interrupted. Aborting exectution.");
|
||||
} finally {
|
||||
dataStore.flush();
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
*
|
||||
* @param query
|
||||
* @return
|
||||
*/
|
||||
public Result get(final Query query) {
|
||||
return get(query, Grouping.NO_GROUPING);
|
||||
}
|
||||
/**
|
||||
*
|
||||
* @param query
|
||||
* @return
|
||||
*/
|
||||
public Result get(final Query query) {
|
||||
return get(query, Grouping.NO_GROUPING);
|
||||
}
|
||||
|
||||
public List<PdbFile> getFilesForQuery(final Query query) {
|
||||
return dataStore.getFilesForQuery(query);
|
||||
}
|
||||
public List<PdbFile> getFilesForQuery(final Query query) {
|
||||
return dataStore.getFilesForQuery(query);
|
||||
}
|
||||
|
||||
/**
|
||||
* Return the entries as an unbound, ordered and non-parallel stream.
|
||||
*
|
||||
* @param query
|
||||
* @param groupBy the tag to group by
|
||||
* @return {@link Result}
|
||||
*/
|
||||
public Result get(final Query query, final List<String> groupBy) {
|
||||
final long start = System.nanoTime();
|
||||
final List<PdbFile> pdbFiles = dataStore.getFilesForQuery(query);
|
||||
/**
|
||||
* Return the entries as an unbound, ordered and non-parallel stream.
|
||||
*
|
||||
* @param query
|
||||
* @param groupBy the tag to group by
|
||||
* @return {@link Result}
|
||||
*/
|
||||
public Result get(final Query query, final List<String> groupBy) {
|
||||
final long start = System.nanoTime();
|
||||
final List<PdbFile> pdbFiles = dataStore.getFilesForQuery(query);
|
||||
|
||||
final Grouping grouping = Grouping.groupBy(pdbFiles, groupBy);
|
||||
final Grouping grouping = Grouping.groupBy(pdbFiles, groupBy);
|
||||
|
||||
final Result result = toResult(grouping);
|
||||
METRICS_LOGGER.debug("query execution took: " + (System.nanoTime() - start) / 1_000_000.0 + "ms: " + query
|
||||
+ " (" + groupBy + "): files found: " + pdbFiles.size());
|
||||
return result;
|
||||
}
|
||||
final Result result = toResult(grouping);
|
||||
METRICS_LOGGER.debug("query execution took: " + (System.nanoTime() - start) / 1_000_000.0 + "ms: " + query
|
||||
+ " (" + groupBy + "): files found: " + pdbFiles.size());
|
||||
return result;
|
||||
}
|
||||
|
||||
private Result toResult(final Grouping grouping) {
|
||||
final List<GroupResult> groupResults = new ArrayList<>();
|
||||
for (final Group group : grouping.getGroups()) {
|
||||
final Stream<LongList> stream = PdbFile.toStream(group.getFiles(), dataStore.getDiskStorage());
|
||||
final GroupResult groupResult = new GroupResult(stream, group.getTags());
|
||||
groupResults.add(groupResult);
|
||||
}
|
||||
final Result result = new Result(groupResults);
|
||||
return result;
|
||||
}
|
||||
private Result toResult(final Grouping grouping) {
|
||||
final List<GroupResult> groupResults = new ArrayList<>();
|
||||
for (final Group group : grouping.getGroups()) {
|
||||
final Stream<LongList> stream = PdbFile.toStream(group.getFiles(), dataStore.getDiskStorage());
|
||||
final GroupResult groupResult = new GroupResult(stream, group.getTags());
|
||||
groupResults.add(groupResult);
|
||||
}
|
||||
final Result result = new Result(groupResults);
|
||||
return result;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void close() {
|
||||
try {
|
||||
dataStore.close();
|
||||
} catch (final Exception e) {
|
||||
LOGGER.error("failed to close PerformanceDB", e);
|
||||
}
|
||||
}
|
||||
@Override
|
||||
public void close() {
|
||||
try {
|
||||
dataStore.close();
|
||||
} catch (final Exception e) {
|
||||
LOGGER.error("failed to close PerformanceDB", e);
|
||||
}
|
||||
}
|
||||
|
||||
public List<Proposal> autocomplete(final QueryWithCaretMarker query) {
|
||||
public List<Proposal> autocomplete(final QueryWithCaretMarker query) {
|
||||
|
||||
return dataStore.propose(query);
|
||||
}
|
||||
return dataStore.propose(query);
|
||||
}
|
||||
|
||||
public List<String> getFields(final DateTimeRange dateRange) {
|
||||
public List<String> getFields(final DateTimeRange dateRange) {
|
||||
|
||||
final List<String> fields = dataStore.getAvailableFields(dateRange);
|
||||
final List<String> fields = dataStore.getAvailableFields(dateRange);
|
||||
|
||||
return fields;
|
||||
}
|
||||
return fields;
|
||||
}
|
||||
|
||||
public PartitionDiskStore getDataStore() {
|
||||
return dataStore.getDiskStorage();
|
||||
}
|
||||
public PartitionDiskStore getDataStore() {
|
||||
return dataStore.getDiskStorage();
|
||||
}
|
||||
}
|
||||
|
||||
@@ -5,18 +5,18 @@ import java.util.Comparator;
|
||||
import org.lucares.pdb.api.Entry;
|
||||
|
||||
public class EntryByDateComparator implements Comparator<Entry> {
|
||||
public static final Comparator<Entry> INSTANCE = new EntryByDateComparator();
|
||||
public static final Comparator<Entry> INSTANCE = new EntryByDateComparator();
|
||||
|
||||
@Override
|
||||
public int compare(final Entry o1, final Entry o2) {
|
||||
@Override
|
||||
public int compare(final Entry o1, final Entry o2) {
|
||||
|
||||
long result = o1.getEpochMilli() - o2.getEpochMilli();
|
||||
long result = o1.getEpochMilli() - o2.getEpochMilli();
|
||||
|
||||
if (result == 0) {
|
||||
result = o1.getValue() - o2.getValue();
|
||||
}
|
||||
if (result == 0) {
|
||||
result = o1.getValue() - o2.getValue();
|
||||
}
|
||||
|
||||
return result < 0 ? -1 : (result == 0 ? 0 : 1);
|
||||
}
|
||||
return result < 0 ? -1 : (result == 0 ? 0 : 1);
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -29,331 +29,331 @@ import org.testng.annotations.Test;
|
||||
@Test
|
||||
public class PerformanceDbTest {
|
||||
|
||||
private Path dataDirectory;
|
||||
private Path dataDirectory;
|
||||
|
||||
@BeforeMethod
|
||||
public void beforeMethod() throws IOException {
|
||||
dataDirectory = Files.createTempDirectory("pdb");
|
||||
}
|
||||
@BeforeMethod
|
||||
public void beforeMethod() throws IOException {
|
||||
dataDirectory = Files.createTempDirectory("pdb");
|
||||
}
|
||||
|
||||
@AfterMethod
|
||||
public void afterMethod() throws IOException {
|
||||
org.lucares.utils.file.FileUtils.delete(dataDirectory);
|
||||
}
|
||||
@AfterMethod
|
||||
public void afterMethod() throws IOException {
|
||||
org.lucares.utils.file.FileUtils.delete(dataDirectory);
|
||||
}
|
||||
|
||||
public void testInsertRead() throws Exception {
|
||||
public void testInsertRead() throws Exception {
|
||||
|
||||
try (PerformanceDb db = new PerformanceDb(dataDirectory)) {
|
||||
final OffsetDateTime nowInUtc = DateUtils.nowInUtc();
|
||||
final long date = nowInUtc.toInstant().toEpochMilli();
|
||||
final long value = 1;
|
||||
final Tags tags = Tags.createAndAddToDictionary("myKey", "myValue");
|
||||
db.putEntry(new Entry(date, value, tags));
|
||||
|
||||
final Result result = db.get(Query.createQuery(tags, DateTimeRange.ofDay(nowInUtc)));
|
||||
final LongList stream = result.singleGroup().flatMap();
|
||||
|
||||
Assert.assertEquals(stream.size(), 2);
|
||||
|
||||
Assert.assertEquals(stream.get(0), date);
|
||||
Assert.assertEquals(stream.get(1), value);
|
||||
}
|
||||
}
|
||||
|
||||
public void testInsertIntoMultipleFilesRead() throws Exception {
|
||||
|
||||
try (PerformanceDb db = new PerformanceDb(dataDirectory)) {
|
||||
final DateTimeRange dateRange = new DateTimeRange(DateUtils.getDate(2016, 11, 1, 10, 0, 0),
|
||||
DateUtils.getDate(2016, 11, 2, 12, 34, 56));
|
||||
final long dayOne = dateRange.getStartEpochMilli();
|
||||
final long dayTwo = dateRange.getEndEpochMilli();
|
||||
final long valueOne = 1;
|
||||
final long valueTwo = 2;
|
||||
final Tags tags = Tags.createAndAddToDictionary("myKey", "myValue");
|
||||
|
||||
db.putEntry(new Entry(dayOne, valueOne, tags));
|
||||
db.putEntry(new Entry(dayTwo, valueTwo, tags));
|
||||
|
||||
final LongList stream = db.get(Query.createQuery(tags, dateRange)).singleGroup().flatMap();
|
||||
|
||||
Assert.assertEquals(stream.size(), 4);
|
||||
|
||||
Assert.assertEquals(stream.get(0), dayOne);
|
||||
Assert.assertEquals(stream.get(1), valueOne);
|
||||
Assert.assertEquals(stream.get(2), dayTwo);
|
||||
Assert.assertEquals(stream.get(3), valueTwo);
|
||||
}
|
||||
}
|
||||
|
||||
private List<Entry> generateEntries(final DateTimeRange dateRange, final long n, final int addToDate,
|
||||
final Tags tags) {
|
||||
final List<Entry> result = new ArrayList<>();
|
||||
final long differenceInMs = dateRange.duration().toMillis() / n;
|
||||
long currentTime = dateRange.getStartEpochMilli();
|
||||
|
||||
for (long i = 0; i < n; i++) {
|
||||
final long value = ThreadLocalRandom.current().nextInt(0, Integer.MAX_VALUE);
|
||||
final long date = OffsetDateTime.ofInstant(Instant.ofEpochMilli(currentTime + addToDate), ZoneOffset.UTC)
|
||||
.toInstant().toEpochMilli();
|
||||
result.add(new Entry(date, value, tags));
|
||||
|
||||
currentTime += differenceInMs;
|
||||
}
|
||||
return result;
|
||||
}
|
||||
|
||||
@DataProvider
|
||||
public Object[][] providerAppendToExistingFile() throws Exception {
|
||||
return new Object[][] { //
|
||||
{ 2 }, //
|
||||
{ 100 }, //
|
||||
{ 500 }, //
|
||||
};
|
||||
}
|
||||
|
||||
@Test(dataProvider = "providerAppendToExistingFile")
|
||||
public void testAppendToExistingFile(final long numberOfEntries) throws Exception {
|
||||
|
||||
try (PerformanceDb db = new PerformanceDb(dataDirectory)) {
|
||||
|
||||
final int year = 2016;
|
||||
final int month = 1;
|
||||
final int day = 2;
|
||||
|
||||
final DateTimeRange timeRange = DateTimeRange.ofDay(DateUtils.getDate(year, month, day, 1, 1, 1));
|
||||
|
||||
final Tags tags = Tags.createAndAddToDictionary("myKey", "one");
|
||||
final List<Entry> entries = generateEntries(timeRange, numberOfEntries, 0, tags);
|
||||
|
||||
printEntries(entries, "");
|
||||
|
||||
db.putEntries(entries);
|
||||
|
||||
final LongList actualEntries = db.get(Query.createQuery(tags, timeRange)).singleGroup().flatMap();
|
||||
Assert.assertEquals(actualEntries.size(), entries.size() * 2);
|
||||
|
||||
for (int i = 0; i < entries.size(); i++) {
|
||||
final Entry entry = entries.get(i);
|
||||
final long epochMilli = entry.getEpochMilli();
|
||||
final long value = entry.getValue();
|
||||
|
||||
Assert.assertEquals(actualEntries.get(i * 2), epochMilli);
|
||||
Assert.assertEquals(actualEntries.get(i * 2 + 1), value);
|
||||
}
|
||||
|
||||
}
|
||||
}
|
||||
|
||||
@DataProvider
|
||||
public Object[][] providerAppendToExistingFileWithRestart() throws Exception {
|
||||
return new Object[][] { //
|
||||
{ 2 }, //
|
||||
{ 100 }, //
|
||||
{ 500 }, //
|
||||
};
|
||||
}
|
||||
|
||||
@Test(dataProvider = "providerAppendToExistingFileWithRestart")
|
||||
public void testAppendToExistingFileWithRestart(final long numberOfEntries) throws Exception {
|
||||
final Tags tags;
|
||||
final List<Entry> expected = new ArrayList<>();
|
||||
try (PerformanceDb db = new PerformanceDb(dataDirectory)) {
|
||||
|
||||
final int year = 2016;
|
||||
final int month = 1;
|
||||
final int day = 2;
|
||||
|
||||
tags = Tags.createAndAddToDictionary("myKey", "one");
|
||||
final DateTimeRange timeRange = DateTimeRange.ofDay(DateUtils.getDate(year, month, day, 1, 1, 1));
|
||||
|
||||
final List<Entry> entries = generateEntries(timeRange, numberOfEntries, 0, tags);
|
||||
db.putEntries(entries);
|
||||
expected.addAll(entries);
|
||||
}
|
||||
|
||||
try (PerformanceDb db = new PerformanceDb(dataDirectory)) {
|
||||
final int year = 2016;
|
||||
final int month = 1;
|
||||
final int day = 3;
|
||||
|
||||
final DateTimeRange timeRange = DateTimeRange.ofDay(DateUtils.getDate(year, month, day, 1, 1, 1));
|
||||
final List<Entry> entries = generateEntries(timeRange, numberOfEntries, 0, tags);
|
||||
db.putEntries(entries);
|
||||
expected.addAll(entries);
|
||||
|
||||
final LongList actualEntries = db.get(Query.createQuery(tags, timeRange)).singleGroup().flatMap();
|
||||
Assert.assertEquals(actualEntries.size(), expected.size() * 2);
|
||||
|
||||
Assert.assertEquals(actualEntries, toExpectedValues(expected));
|
||||
}
|
||||
}
|
||||
|
||||
public void testInsertIntoMultipleFilesWithDifferentTags() throws Exception {
|
||||
|
||||
try (PerformanceDb db = new PerformanceDb(dataDirectory)) {
|
||||
final OffsetDateTime from = DateUtils.getDate(2016, 1, 1, 00, 00, 00);
|
||||
final OffsetDateTime to = DateUtils.getDate(2016, 1, 1, 23, 59, 50);
|
||||
|
||||
final DateTimeRange timeRange = new DateTimeRange(from, to);
|
||||
final DateTimeRange dateRange = new DateTimeRange(from, to);
|
||||
final long numberOfEntries = timeRange.duration().toHours();
|
||||
|
||||
final Tags tagsCommon = Tags.createAndAddToDictionary("commonKey", "commonValue");
|
||||
final Tags tagsOne = Tags.createAndAddToDictionary("myKey", "one", "commonKey", "commonValue");
|
||||
final List<Entry> entriesOne = generateEntries(timeRange, numberOfEntries, 1, tagsOne);
|
||||
db.putEntries(entriesOne);
|
||||
printEntries(entriesOne, "one");
|
||||
|
||||
final Tags tagsTwo = Tags.createAndAddToDictionary("myKey", "two", "commonKey", "commonValue");
|
||||
final List<Entry> entriesTwo = generateEntries(timeRange, numberOfEntries, 2, tagsTwo);
|
||||
printEntries(entriesTwo, "two");
|
||||
db.putEntries(entriesTwo);
|
||||
|
||||
final Tags tagsThree = Tags.createAndAddToDictionary("myKey", "three", "commonKey", "commonValue");
|
||||
final List<Entry> entriesThree = generateEntries(timeRange, numberOfEntries, 3, tagsThree);
|
||||
printEntries(entriesThree, "three");
|
||||
db.putEntries(entriesThree);
|
||||
|
||||
final LongList actualEntriesOne = db.get(Query.createQuery(tagsOne, dateRange)).singleGroup().flatMap();
|
||||
Assert.assertEquals(actualEntriesOne, toExpectedValues(entriesOne));
|
||||
|
||||
final LongList actualEntriesTwo = db.get(Query.createQuery(tagsTwo, dateRange)).singleGroup().flatMap();
|
||||
Assert.assertEquals(actualEntriesTwo, toExpectedValues(entriesTwo));
|
||||
|
||||
final LongList actualEntriesThree = db.get(Query.createQuery(tagsThree, dateRange)).singleGroup().flatMap();
|
||||
Assert.assertEquals(actualEntriesThree, toExpectedValues(entriesThree));
|
||||
|
||||
final LongList actualEntriesAll = db.get(Query.createQuery(tagsCommon, dateRange)).singleGroup().flatMap();
|
||||
final List<Entry> expectedAll = CollectionUtils.collate(entriesOne,
|
||||
CollectionUtils.collate(entriesTwo, entriesThree, EntryByDateComparator.INSTANCE),
|
||||
EntryByDateComparator.INSTANCE);
|
||||
final LongList expectedValues = toExpectedValues(expectedAll);
|
||||
|
||||
actualEntriesAll.sort();
|
||||
expectedValues.sort();
|
||||
|
||||
Assert.assertEquals(actualEntriesAll, expectedValues);
|
||||
}
|
||||
}
|
||||
|
||||
public void testGroupBySingleField() throws Exception {
|
||||
try (PerformanceDb db = new PerformanceDb(dataDirectory)) {
|
||||
final OffsetDateTime from = DateUtils.getDate(2016, 1, 1, 00, 00, 00);
|
||||
final OffsetDateTime to = DateUtils.getDate(2016, 1, 1, 23, 59, 50);
|
||||
|
||||
final DateTimeRange timeRange = new DateTimeRange(from, to);
|
||||
final long numberOfEntries = timeRange.duration().toHours();
|
||||
|
||||
final String key = "myKey";
|
||||
final Tags tagsOne = Tags.createAndAddToDictionary(key, "one", "commonKey", "commonValue");
|
||||
final Tags tagsTwo = Tags.createAndAddToDictionary(key, "two", "commonKey", "commonValue");
|
||||
final Tags tagsThree = Tags.createAndAddToDictionary("commonKey", "commonValue");
|
||||
final LongList entriesOne = storeEntries(db, timeRange, numberOfEntries, tagsOne, 1);
|
||||
final LongList entriesTwo = storeEntries(db, timeRange, numberOfEntries, tagsTwo, 2);
|
||||
final LongList entriesThree = storeEntries(db, timeRange, numberOfEntries, tagsThree, 3);
|
||||
|
||||
final Result result = db.get(Query.createQuery("commonKey=commonValue", timeRange), Arrays.asList(key));
|
||||
|
||||
final List<GroupResult> groups = result.getGroups();
|
||||
|
||||
for (final GroupResult groupResult : groups) {
|
||||
final Tags groupedBy = groupResult.getGroupedBy();
|
||||
|
||||
if (groupedBy.equals(Tags.createAndAddToDictionary(key, "one"))) {
|
||||
Assert.assertEquals(groupResult.flatMap(), entriesOne);
|
||||
} else if (groupedBy.equals(Tags.createAndAddToDictionary(key, "two"))) {
|
||||
|
||||
Assert.assertEquals(groupResult.flatMap(), entriesTwo);
|
||||
} else if (groupedBy.isEmpty()) {
|
||||
Assert.assertEquals(groupResult.flatMap(), entriesThree);
|
||||
} else {
|
||||
Assert.fail("unexpected group: " + groupResult.getGroupedBy());
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
public void testGroupByMultipleFields() throws Exception {
|
||||
try (PerformanceDb db = new PerformanceDb(dataDirectory)) {
|
||||
final OffsetDateTime from = DateUtils.getDate(2016, 1, 1, 00, 00, 00);
|
||||
final OffsetDateTime to = DateUtils.getDate(2016, 1, 1, 23, 59, 50);
|
||||
|
||||
final DateTimeRange timeRange = new DateTimeRange(from, to);
|
||||
final long numberOfEntries = timeRange.duration().toHours();
|
||||
|
||||
final String key1 = "myKey1";
|
||||
final String key2 = "myKey2";
|
||||
final Tags tagsOne = Tags.createAndAddToDictionary(key1, "one", key2, "aaa", "commonKey", "commonValue");
|
||||
final Tags tagsTwoA = Tags.createAndAddToDictionary(key1, "two", key2, "bbb", "commonKey", "commonValue");
|
||||
final Tags tagsTwoB = Tags.createAndAddToDictionary(key1, "two", key2, "bbb", "commonKey", "commonValue");
|
||||
final Tags tagsThree = Tags.createAndAddToDictionary(key1, "three", "commonKey", "commonValue");
|
||||
|
||||
final LongList entriesOne = storeEntries(db, timeRange, numberOfEntries, tagsOne, 1);
|
||||
final LongList entriesTwo = storeEntries(db, timeRange, numberOfEntries, tagsTwoA, 2);
|
||||
entriesTwo.addAll(storeEntries(db, timeRange, numberOfEntries, tagsTwoB, 3));
|
||||
final LongList entriesThree = storeEntries(db, timeRange, numberOfEntries, tagsThree, 4);
|
||||
|
||||
final Result result = db.get(Query.createQuery("commonKey=commonValue", timeRange),
|
||||
Arrays.asList(key1, key2));
|
||||
|
||||
final List<GroupResult> groups = result.getGroups();
|
||||
|
||||
for (final GroupResult groupResult : groups) {
|
||||
final Tags groupedBy = groupResult.getGroupedBy();
|
||||
|
||||
if (groupedBy.equals(Tags.createAndAddToDictionary(key1, "one", key2, "aaa"))) {
|
||||
Assert.assertEquals(groupResult.flatMap(), entriesOne);
|
||||
} else if (groupedBy.equals(Tags.createAndAddToDictionary(key1, "two", key2, "bbb"))) {
|
||||
// there is no defined order of the entries.
|
||||
// eventually we might return them in ascending order, but
|
||||
// that is not yet implemented
|
||||
final LongList actualEntries = groupResult.flatMap();
|
||||
|
||||
entriesTwo.sort();
|
||||
actualEntries.sort();
|
||||
|
||||
Assert.assertEquals(actualEntries, entriesTwo);
|
||||
} else if (groupedBy.equals(Tags.createAndAddToDictionary(key1, "three"))) {
|
||||
Assert.assertEquals(groupResult.flatMap(), entriesThree);
|
||||
} else {
|
||||
Assert.fail("unexpected group: " + groupedBy);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private LongList storeEntries(final PerformanceDb performanceDb, final DateTimeRange timeRange,
|
||||
final long numberOfEntries, final Tags tags, final int addToDate) {
|
||||
final List<Entry> entries = generateEntries(timeRange, numberOfEntries, addToDate, tags);
|
||||
performanceDb.putEntries(entries);
|
||||
|
||||
final LongList result = new LongList();
|
||||
|
||||
for (final Entry entry : entries) {
|
||||
result.add(entry.getEpochMilli());
|
||||
result.add(entry.getValue());
|
||||
}
|
||||
|
||||
return result;
|
||||
}
|
||||
|
||||
private void printEntries(final List<Entry> entriesOne, final String label) {
|
||||
|
||||
int index = 0;
|
||||
for (final Entry entry : entriesOne) {
|
||||
System.out.printf("%4d %s %d (%s)\n", index, entry.getEpochMilli(), entry.getValue(), label);
|
||||
index++;
|
||||
}
|
||||
}
|
||||
|
||||
private LongList toExpectedValues(final List<Entry> entries) {
|
||||
|
||||
final LongList result = new LongList();
|
||||
for (final Entry entry : entries) {
|
||||
result.add(entry.getEpochMilli());
|
||||
result.add(entry.getValue());
|
||||
}
|
||||
|
||||
return result;
|
||||
}
|
||||
try (PerformanceDb db = new PerformanceDb(dataDirectory)) {
|
||||
final OffsetDateTime nowInUtc = DateUtils.nowInUtc();
|
||||
final long date = nowInUtc.toInstant().toEpochMilli();
|
||||
final long value = 1;
|
||||
final Tags tags = Tags.createAndAddToDictionary("myKey", "myValue");
|
||||
db.putEntry(new Entry(date, value, tags));
|
||||
|
||||
final Result result = db.get(Query.createQuery(tags, DateTimeRange.ofDay(nowInUtc)));
|
||||
final LongList stream = result.singleGroup().flatMap();
|
||||
|
||||
Assert.assertEquals(stream.size(), 2);
|
||||
|
||||
Assert.assertEquals(stream.get(0), date);
|
||||
Assert.assertEquals(stream.get(1), value);
|
||||
}
|
||||
}
|
||||
|
||||
public void testInsertIntoMultipleFilesRead() throws Exception {
|
||||
|
||||
try (PerformanceDb db = new PerformanceDb(dataDirectory)) {
|
||||
final DateTimeRange dateRange = new DateTimeRange(DateUtils.getDate(2016, 11, 1, 10, 0, 0),
|
||||
DateUtils.getDate(2016, 11, 2, 12, 34, 56));
|
||||
final long dayOne = dateRange.getStartEpochMilli();
|
||||
final long dayTwo = dateRange.getEndEpochMilli();
|
||||
final long valueOne = 1;
|
||||
final long valueTwo = 2;
|
||||
final Tags tags = Tags.createAndAddToDictionary("myKey", "myValue");
|
||||
|
||||
db.putEntry(new Entry(dayOne, valueOne, tags));
|
||||
db.putEntry(new Entry(dayTwo, valueTwo, tags));
|
||||
|
||||
final LongList stream = db.get(Query.createQuery(tags, dateRange)).singleGroup().flatMap();
|
||||
|
||||
Assert.assertEquals(stream.size(), 4);
|
||||
|
||||
Assert.assertEquals(stream.get(0), dayOne);
|
||||
Assert.assertEquals(stream.get(1), valueOne);
|
||||
Assert.assertEquals(stream.get(2), dayTwo);
|
||||
Assert.assertEquals(stream.get(3), valueTwo);
|
||||
}
|
||||
}
|
||||
|
||||
private List<Entry> generateEntries(final DateTimeRange dateRange, final long n, final int addToDate,
|
||||
final Tags tags) {
|
||||
final List<Entry> result = new ArrayList<>();
|
||||
final long differenceInMs = dateRange.duration().toMillis() / n;
|
||||
long currentTime = dateRange.getStartEpochMilli();
|
||||
|
||||
for (long i = 0; i < n; i++) {
|
||||
final long value = ThreadLocalRandom.current().nextInt(0, Integer.MAX_VALUE);
|
||||
final long date = OffsetDateTime.ofInstant(Instant.ofEpochMilli(currentTime + addToDate), ZoneOffset.UTC)
|
||||
.toInstant().toEpochMilli();
|
||||
result.add(new Entry(date, value, tags));
|
||||
|
||||
currentTime += differenceInMs;
|
||||
}
|
||||
return result;
|
||||
}
|
||||
|
||||
@DataProvider
|
||||
public Object[][] providerAppendToExistingFile() throws Exception {
|
||||
return new Object[][] { //
|
||||
{ 2 }, //
|
||||
{ 100 }, //
|
||||
{ 500 }, //
|
||||
};
|
||||
}
|
||||
|
||||
@Test(dataProvider = "providerAppendToExistingFile")
|
||||
public void testAppendToExistingFile(final long numberOfEntries) throws Exception {
|
||||
|
||||
try (PerformanceDb db = new PerformanceDb(dataDirectory)) {
|
||||
|
||||
final int year = 2016;
|
||||
final int month = 1;
|
||||
final int day = 2;
|
||||
|
||||
final DateTimeRange timeRange = DateTimeRange.ofDay(DateUtils.getDate(year, month, day, 1, 1, 1));
|
||||
|
||||
final Tags tags = Tags.createAndAddToDictionary("myKey", "one");
|
||||
final List<Entry> entries = generateEntries(timeRange, numberOfEntries, 0, tags);
|
||||
|
||||
printEntries(entries, "");
|
||||
|
||||
db.putEntries(entries);
|
||||
|
||||
final LongList actualEntries = db.get(Query.createQuery(tags, timeRange)).singleGroup().flatMap();
|
||||
Assert.assertEquals(actualEntries.size(), entries.size() * 2);
|
||||
|
||||
for (int i = 0; i < entries.size(); i++) {
|
||||
final Entry entry = entries.get(i);
|
||||
final long epochMilli = entry.getEpochMilli();
|
||||
final long value = entry.getValue();
|
||||
|
||||
Assert.assertEquals(actualEntries.get(i * 2), epochMilli);
|
||||
Assert.assertEquals(actualEntries.get(i * 2 + 1), value);
|
||||
}
|
||||
|
||||
}
|
||||
}
|
||||
|
||||
@DataProvider
|
||||
public Object[][] providerAppendToExistingFileWithRestart() throws Exception {
|
||||
return new Object[][] { //
|
||||
{ 2 }, //
|
||||
{ 100 }, //
|
||||
{ 500 }, //
|
||||
};
|
||||
}
|
||||
|
||||
@Test(dataProvider = "providerAppendToExistingFileWithRestart")
|
||||
public void testAppendToExistingFileWithRestart(final long numberOfEntries) throws Exception {
|
||||
final Tags tags;
|
||||
final List<Entry> expected = new ArrayList<>();
|
||||
try (PerformanceDb db = new PerformanceDb(dataDirectory)) {
|
||||
|
||||
final int year = 2016;
|
||||
final int month = 1;
|
||||
final int day = 2;
|
||||
|
||||
tags = Tags.createAndAddToDictionary("myKey", "one");
|
||||
final DateTimeRange timeRange = DateTimeRange.ofDay(DateUtils.getDate(year, month, day, 1, 1, 1));
|
||||
|
||||
final List<Entry> entries = generateEntries(timeRange, numberOfEntries, 0, tags);
|
||||
db.putEntries(entries);
|
||||
expected.addAll(entries);
|
||||
}
|
||||
|
||||
try (PerformanceDb db = new PerformanceDb(dataDirectory)) {
|
||||
final int year = 2016;
|
||||
final int month = 1;
|
||||
final int day = 3;
|
||||
|
||||
final DateTimeRange timeRange = DateTimeRange.ofDay(DateUtils.getDate(year, month, day, 1, 1, 1));
|
||||
final List<Entry> entries = generateEntries(timeRange, numberOfEntries, 0, tags);
|
||||
db.putEntries(entries);
|
||||
expected.addAll(entries);
|
||||
|
||||
final LongList actualEntries = db.get(Query.createQuery(tags, timeRange)).singleGroup().flatMap();
|
||||
Assert.assertEquals(actualEntries.size(), expected.size() * 2);
|
||||
|
||||
Assert.assertEquals(actualEntries, toExpectedValues(expected));
|
||||
}
|
||||
}
|
||||
|
||||
public void testInsertIntoMultipleFilesWithDifferentTags() throws Exception {
|
||||
|
||||
try (PerformanceDb db = new PerformanceDb(dataDirectory)) {
|
||||
final OffsetDateTime from = DateUtils.getDate(2016, 1, 1, 00, 00, 00);
|
||||
final OffsetDateTime to = DateUtils.getDate(2016, 1, 1, 23, 59, 50);
|
||||
|
||||
final DateTimeRange timeRange = new DateTimeRange(from, to);
|
||||
final DateTimeRange dateRange = new DateTimeRange(from, to);
|
||||
final long numberOfEntries = timeRange.duration().toHours();
|
||||
|
||||
final Tags tagsCommon = Tags.createAndAddToDictionary("commonKey", "commonValue");
|
||||
final Tags tagsOne = Tags.createAndAddToDictionary("myKey", "one", "commonKey", "commonValue");
|
||||
final List<Entry> entriesOne = generateEntries(timeRange, numberOfEntries, 1, tagsOne);
|
||||
db.putEntries(entriesOne);
|
||||
printEntries(entriesOne, "one");
|
||||
|
||||
final Tags tagsTwo = Tags.createAndAddToDictionary("myKey", "two", "commonKey", "commonValue");
|
||||
final List<Entry> entriesTwo = generateEntries(timeRange, numberOfEntries, 2, tagsTwo);
|
||||
printEntries(entriesTwo, "two");
|
||||
db.putEntries(entriesTwo);
|
||||
|
||||
final Tags tagsThree = Tags.createAndAddToDictionary("myKey", "three", "commonKey", "commonValue");
|
||||
final List<Entry> entriesThree = generateEntries(timeRange, numberOfEntries, 3, tagsThree);
|
||||
printEntries(entriesThree, "three");
|
||||
db.putEntries(entriesThree);
|
||||
|
||||
final LongList actualEntriesOne = db.get(Query.createQuery(tagsOne, dateRange)).singleGroup().flatMap();
|
||||
Assert.assertEquals(actualEntriesOne, toExpectedValues(entriesOne));
|
||||
|
||||
final LongList actualEntriesTwo = db.get(Query.createQuery(tagsTwo, dateRange)).singleGroup().flatMap();
|
||||
Assert.assertEquals(actualEntriesTwo, toExpectedValues(entriesTwo));
|
||||
|
||||
final LongList actualEntriesThree = db.get(Query.createQuery(tagsThree, dateRange)).singleGroup().flatMap();
|
||||
Assert.assertEquals(actualEntriesThree, toExpectedValues(entriesThree));
|
||||
|
||||
final LongList actualEntriesAll = db.get(Query.createQuery(tagsCommon, dateRange)).singleGroup().flatMap();
|
||||
final List<Entry> expectedAll = CollectionUtils.collate(entriesOne,
|
||||
CollectionUtils.collate(entriesTwo, entriesThree, EntryByDateComparator.INSTANCE),
|
||||
EntryByDateComparator.INSTANCE);
|
||||
final LongList expectedValues = toExpectedValues(expectedAll);
|
||||
|
||||
actualEntriesAll.sort();
|
||||
expectedValues.sort();
|
||||
|
||||
Assert.assertEquals(actualEntriesAll, expectedValues);
|
||||
}
|
||||
}
|
||||
|
||||
public void testGroupBySingleField() throws Exception {
|
||||
try (PerformanceDb db = new PerformanceDb(dataDirectory)) {
|
||||
final OffsetDateTime from = DateUtils.getDate(2016, 1, 1, 00, 00, 00);
|
||||
final OffsetDateTime to = DateUtils.getDate(2016, 1, 1, 23, 59, 50);
|
||||
|
||||
final DateTimeRange timeRange = new DateTimeRange(from, to);
|
||||
final long numberOfEntries = timeRange.duration().toHours();
|
||||
|
||||
final String key = "myKey";
|
||||
final Tags tagsOne = Tags.createAndAddToDictionary(key, "one", "commonKey", "commonValue");
|
||||
final Tags tagsTwo = Tags.createAndAddToDictionary(key, "two", "commonKey", "commonValue");
|
||||
final Tags tagsThree = Tags.createAndAddToDictionary("commonKey", "commonValue");
|
||||
final LongList entriesOne = storeEntries(db, timeRange, numberOfEntries, tagsOne, 1);
|
||||
final LongList entriesTwo = storeEntries(db, timeRange, numberOfEntries, tagsTwo, 2);
|
||||
final LongList entriesThree = storeEntries(db, timeRange, numberOfEntries, tagsThree, 3);
|
||||
|
||||
final Result result = db.get(Query.createQuery("commonKey=commonValue", timeRange), Arrays.asList(key));
|
||||
|
||||
final List<GroupResult> groups = result.getGroups();
|
||||
|
||||
for (final GroupResult groupResult : groups) {
|
||||
final Tags groupedBy = groupResult.getGroupedBy();
|
||||
|
||||
if (groupedBy.equals(Tags.createAndAddToDictionary(key, "one"))) {
|
||||
Assert.assertEquals(groupResult.flatMap(), entriesOne);
|
||||
} else if (groupedBy.equals(Tags.createAndAddToDictionary(key, "two"))) {
|
||||
|
||||
Assert.assertEquals(groupResult.flatMap(), entriesTwo);
|
||||
} else if (groupedBy.isEmpty()) {
|
||||
Assert.assertEquals(groupResult.flatMap(), entriesThree);
|
||||
} else {
|
||||
Assert.fail("unexpected group: " + groupResult.getGroupedBy());
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
public void testGroupByMultipleFields() throws Exception {
|
||||
try (PerformanceDb db = new PerformanceDb(dataDirectory)) {
|
||||
final OffsetDateTime from = DateUtils.getDate(2016, 1, 1, 00, 00, 00);
|
||||
final OffsetDateTime to = DateUtils.getDate(2016, 1, 1, 23, 59, 50);
|
||||
|
||||
final DateTimeRange timeRange = new DateTimeRange(from, to);
|
||||
final long numberOfEntries = timeRange.duration().toHours();
|
||||
|
||||
final String key1 = "myKey1";
|
||||
final String key2 = "myKey2";
|
||||
final Tags tagsOne = Tags.createAndAddToDictionary(key1, "one", key2, "aaa", "commonKey", "commonValue");
|
||||
final Tags tagsTwoA = Tags.createAndAddToDictionary(key1, "two", key2, "bbb", "commonKey", "commonValue");
|
||||
final Tags tagsTwoB = Tags.createAndAddToDictionary(key1, "two", key2, "bbb", "commonKey", "commonValue");
|
||||
final Tags tagsThree = Tags.createAndAddToDictionary(key1, "three", "commonKey", "commonValue");
|
||||
|
||||
final LongList entriesOne = storeEntries(db, timeRange, numberOfEntries, tagsOne, 1);
|
||||
final LongList entriesTwo = storeEntries(db, timeRange, numberOfEntries, tagsTwoA, 2);
|
||||
entriesTwo.addAll(storeEntries(db, timeRange, numberOfEntries, tagsTwoB, 3));
|
||||
final LongList entriesThree = storeEntries(db, timeRange, numberOfEntries, tagsThree, 4);
|
||||
|
||||
final Result result = db.get(Query.createQuery("commonKey=commonValue", timeRange),
|
||||
Arrays.asList(key1, key2));
|
||||
|
||||
final List<GroupResult> groups = result.getGroups();
|
||||
|
||||
for (final GroupResult groupResult : groups) {
|
||||
final Tags groupedBy = groupResult.getGroupedBy();
|
||||
|
||||
if (groupedBy.equals(Tags.createAndAddToDictionary(key1, "one", key2, "aaa"))) {
|
||||
Assert.assertEquals(groupResult.flatMap(), entriesOne);
|
||||
} else if (groupedBy.equals(Tags.createAndAddToDictionary(key1, "two", key2, "bbb"))) {
|
||||
// there is no defined order of the entries.
|
||||
// eventually we might return them in ascending order, but
|
||||
// that is not yet implemented
|
||||
final LongList actualEntries = groupResult.flatMap();
|
||||
|
||||
entriesTwo.sort();
|
||||
actualEntries.sort();
|
||||
|
||||
Assert.assertEquals(actualEntries, entriesTwo);
|
||||
} else if (groupedBy.equals(Tags.createAndAddToDictionary(key1, "three"))) {
|
||||
Assert.assertEquals(groupResult.flatMap(), entriesThree);
|
||||
} else {
|
||||
Assert.fail("unexpected group: " + groupedBy);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private LongList storeEntries(final PerformanceDb performanceDb, final DateTimeRange timeRange,
|
||||
final long numberOfEntries, final Tags tags, final int addToDate) {
|
||||
final List<Entry> entries = generateEntries(timeRange, numberOfEntries, addToDate, tags);
|
||||
performanceDb.putEntries(entries);
|
||||
|
||||
final LongList result = new LongList();
|
||||
|
||||
for (final Entry entry : entries) {
|
||||
result.add(entry.getEpochMilli());
|
||||
result.add(entry.getValue());
|
||||
}
|
||||
|
||||
return result;
|
||||
}
|
||||
|
||||
private void printEntries(final List<Entry> entriesOne, final String label) {
|
||||
|
||||
int index = 0;
|
||||
for (final Entry entry : entriesOne) {
|
||||
System.out.printf("%4d %s %d (%s)\n", index, entry.getEpochMilli(), entry.getValue(), label);
|
||||
index++;
|
||||
}
|
||||
}
|
||||
|
||||
private LongList toExpectedValues(final List<Entry> entries) {
|
||||
|
||||
final LongList result = new LongList();
|
||||
for (final Entry entry : entries) {
|
||||
result.add(entry.getEpochMilli());
|
||||
result.add(entry.getValue());
|
||||
}
|
||||
|
||||
return result;
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user