souffle-haskell-1.0.0: cbits/souffle/ReadStreamCSV.h
/*
* Souffle - A Datalog Compiler
* Copyright (c) 2013, 2014, Oracle and/or its affiliates. All rights reserved
* Licensed under the Universal Permissive License v 1.0 as shown at:
* - https://opensource.org/licenses/UPL
* - <souffle root>/licenses/SOUFFLE-UPL.txt
*/
/************************************************************************
*
* @file ReadStreamCSV.h
*
***********************************************************************/
#pragma once
#include "RamTypes.h"
#include "ReadStream.h"
#include "SymbolTable.h"
#include "utility/ContainerUtil.h"
#include "utility/FileUtil.h"
#include "utility/StringUtil.h"
#ifdef USE_LIBZ
#include "gzfstream.h"
#else
#include <fstream>
#endif
#include <algorithm>
#include <cassert>
#include <cstddef>
#include <cstdint>
#include <iostream>
#include <map>
#include <memory>
#include <sstream>
#include <stdexcept>
#include <string>
#include <vector>
namespace souffle {
class RecordTable;
class ReadStreamCSV : public ReadStream {
public:
ReadStreamCSV(std::istream& file, const std::map<std::string, std::string>& rwOperation,
SymbolTable& symbolTable, RecordTable& recordTable)
: ReadStream(rwOperation, symbolTable, recordTable),
delimiter(getOr(rwOperation, "delimiter", "\t")), file(file), lineNumber(0),
inputMap(getInputColumnMap(rwOperation, arity)) {
while (inputMap.size() < arity) {
int size = static_cast<int>(inputMap.size());
inputMap[size] = size;
}
}
protected:
/**
* Read and return the next tuple.
*
* Returns nullptr if no tuple was readable.
* @return
*/
std::unique_ptr<RamDomain[]> readNextTuple() override {
if (file.eof()) {
return nullptr;
}
std::string line;
std::unique_ptr<RamDomain[]> tuple = std::make_unique<RamDomain[]>(typeAttributes.size());
if (!getline(file, line)) {
return nullptr;
}
// Handle Windows line endings on non-Windows systems
if (!line.empty() && line.back() == '\r') {
line = line.substr(0, line.length() - 1);
}
++lineNumber;
size_t start = 0;
size_t end = 0;
size_t columnsFilled = 0;
for (uint32_t column = 0; columnsFilled < arity; column++) {
size_t charactersRead = 0;
std::string element = nextElement(line, start, end);
if (inputMap.count(column) == 0) {
continue;
}
++columnsFilled;
try {
auto&& ty = typeAttributes.at(inputMap[column]);
switch (ty[0]) {
case 's': {
tuple[inputMap[column]] = symbolTable.unsafeLookup(element);
charactersRead = element.size();
break;
}
case 'r': {
tuple[inputMap[column]] = readRecord(element, ty, 0, &charactersRead);
break;
}
case 'i': {
tuple[inputMap[column]] = RamSignedFromString(element, &charactersRead);
break;
}
case 'u': {
tuple[inputMap[column]] = ramBitCast(readRamUnsigned(element, charactersRead));
break;
}
case 'f': {
tuple[inputMap[column]] = ramBitCast(RamFloatFromString(element, &charactersRead));
break;
}
default: fatal("invalid type attribute: `%c`", ty[0]);
}
// Check if everything was read.
if (charactersRead != element.size()) {
throw std::invalid_argument(
"Expected: " + delimiter + " or \\n. Got: " + element[charactersRead]);
}
} catch (...) {
std::stringstream errorMessage;
errorMessage << "Error converting <" + element + "> in column " << column + 1 << " in line "
<< lineNumber << "; ";
throw std::invalid_argument(errorMessage.str());
}
}
return tuple;
}
/**
* Read an unsigned element. Possible bases are 2, 10, 16
* Base is indicated by the first two chars.
*/
RamUnsigned readRamUnsigned(const std::string& element, size_t& charactersRead) {
// Sanity check
assert(element.size() > 0);
RamSigned value = 0;
// Check prefix and parse the input.
if (isPrefix("0b", element)) {
value = RamUnsignedFromString(element, &charactersRead, 2);
} else if (isPrefix("0x", element)) {
value = RamUnsignedFromString(element, &charactersRead, 16);
} else {
value = RamUnsignedFromString(element, &charactersRead);
}
return value;
}
std::string nextElement(const std::string& line, size_t& start, size_t& end) {
std::string element;
// Handle record/tuple delimiter coincidence.
if (delimiter.find(',') != std::string::npos) {
int record_parens = 0;
size_t next_delimiter = line.find(delimiter, start);
// Find first delimiter after the record.
while (end < std::min(next_delimiter, line.length()) || record_parens != 0) {
// Track the number of parenthesis.
if (line[end] == '[') {
++record_parens;
} else if (line[end] == ']') {
--record_parens;
}
// Check for unbalanced parenthesis.
if (record_parens < 0) {
break;
};
++end;
// Find a next delimiter if the old one is invalid.
// But only if inside the unbalance parenthesis.
if (end == next_delimiter && record_parens != 0) {
next_delimiter = line.find(delimiter, end);
}
}
// Handle the end-of-the-line case where parenthesis are unbalanced.
if (record_parens != 0) {
std::stringstream errorMessage;
errorMessage << "Unbalanced record parenthesis " << lineNumber << "; ";
throw std::invalid_argument(errorMessage.str());
}
} else {
end = std::min(line.find(delimiter, start), line.length());
}
// Check for missing value.
if (start > end) {
std::stringstream errorMessage;
errorMessage << "Values missing in line " << lineNumber << "; ";
throw std::invalid_argument(errorMessage.str());
}
element = line.substr(start, end - start);
start = end + delimiter.size();
return element;
}
std::map<int, int> getInputColumnMap(
const std::map<std::string, std::string>& rwOperation, const unsigned arity_) const {
std::string columnString = getOr(rwOperation, "columns", "");
std::map<int, int> inputColumnMap;
if (!columnString.empty()) {
std::istringstream iss(columnString);
std::string mapping;
int index = 0;
while (std::getline(iss, mapping, ':')) {
inputColumnMap[stoi(mapping)] = index++;
}
if (inputColumnMap.size() < arity_) {
throw std::invalid_argument("Invalid column set was given: <" + columnString + ">");
}
} else {
while (inputColumnMap.size() < arity_) {
int size = static_cast<int>(inputColumnMap.size());
inputColumnMap[size] = size;
}
}
return inputColumnMap;
}
const std::string delimiter;
std::istream& file;
size_t lineNumber;
std::map<int, int> inputMap;
};
class ReadFileCSV : public ReadStreamCSV {
public:
ReadFileCSV(const std::map<std::string, std::string>& rwOperation, SymbolTable& symbolTable,
RecordTable& recordTable)
: ReadStreamCSV(fileHandle, rwOperation, symbolTable, recordTable),
baseName(souffle::baseName(getFileName(rwOperation))),
fileHandle(getFileName(rwOperation), std::ios::in | std::ios::binary) {
if (!fileHandle.is_open()) {
throw std::invalid_argument("Cannot open fact file " + baseName + "\n");
}
// Strip headers if we're using them
if (getOr(rwOperation, "headers", "false") == "true") {
std::string line;
getline(file, line);
}
}
/**
* Read and return the next tuple.
*
* Returns nullptr if no tuple was readable.
* @return
*/
std::unique_ptr<RamDomain[]> readNextTuple() override {
try {
return ReadStreamCSV::readNextTuple();
} catch (std::exception& e) {
std::stringstream errorMessage;
errorMessage << e.what();
errorMessage << "cannot parse fact file " << baseName << "!\n";
throw std::invalid_argument(errorMessage.str());
}
}
~ReadFileCSV() override = default;
protected:
static std::string getFileName(const std::map<std::string, std::string>& rwOperation) {
return getOr(rwOperation, "filename", rwOperation.at("name") + ".facts");
}
std::string baseName;
#ifdef USE_LIBZ
gzfstream::igzfstream fileHandle;
#else
std::ifstream fileHandle;
#endif
};
class ReadCinCSVFactory : public ReadStreamFactory {
public:
std::unique_ptr<ReadStream> getReader(const std::map<std::string, std::string>& rwOperation,
SymbolTable& symbolTable, RecordTable& recordTable) override {
return std::make_unique<ReadStreamCSV>(std::cin, rwOperation, symbolTable, recordTable);
}
const std::string& getName() const override {
static const std::string name = "stdin";
return name;
}
~ReadCinCSVFactory() override = default;
};
class ReadFileCSVFactory : public ReadStreamFactory {
public:
std::unique_ptr<ReadStream> getReader(const std::map<std::string, std::string>& rwOperation,
SymbolTable& symbolTable, RecordTable& recordTable) override {
return std::make_unique<ReadFileCSV>(rwOperation, symbolTable, recordTable);
}
const std::string& getName() const override {
static const std::string name = "file";
return name;
}
~ReadFileCSVFactory() override = default;
};
} /* namespace souffle */