lbug 0.16.1

An in-process property graph database management system built for query speed and scalability
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145

#include "binder/binder.h"
#include "binder/bound_import_database.h"
#include "common/copier_config/csv_reader_config.h"
#include "common/exception/binder.h"
#include "common/file_system/virtual_file_system.h"
#include "main/client_context.h"
#include "parser/copy.h"
#include "parser/parser.h"
#include "parser/port_db.h"
#include <format>

using namespace lbug::common;
using namespace lbug::parser;

namespace lbug {
namespace binder {

static std::string getQueryFromFile(VirtualFileSystem* vfs, const std::string& boundFilePath,
    const std::string& fileName, main::ClientContext* context) {
    auto filePath = vfs->joinPath(boundFilePath, fileName);
    if (!vfs->fileOrPathExists(filePath, context)) {
        if (fileName == PortDBConstants::COPY_FILE_NAME) {
            return "";
        }
        if (fileName == PortDBConstants::INDEX_FILE_NAME) {
            return "";
        }
        throw BinderException(std::format("File {} does not exist.", filePath));
    }
    auto fileInfo = vfs->openFile(filePath, FileOpenFlags(FileFlags::READ_ONLY
#ifdef _WIN32
                                                          | FileFlags::BINARY
#endif
                                                ));
    auto fsize = fileInfo->getFileSize();
    auto buffer = std::make_unique<char[]>(fsize);
    fileInfo->readFile(buffer.get(), fsize);
    return std::string(buffer.get(), fsize);
}

static std::string getColumnNamesToCopy(const CopyFrom& copyFrom) {
    std::string columns = "";
    std::string delimiter = "";
    for (auto& column : copyFrom.getCopyColumnInfo().columnNames) {
        columns += delimiter;
        columns += "`" + column + "`";
        if (delimiter == "") {
            delimiter = ",";
        }
    }
    if (columns.empty()) {
        return columns;
    }
    return std::format("({})", columns);
}

static std::string getCopyFilePath(const std::string& boundFilePath, const std::string& filePath) {
    if (filePath[0] == '/' || (std::isalpha(filePath[0]) && filePath[1] == ':')) {
        // Note:
        // Unix absolute path starts with '/'
        // Windows absolute path starts with "[DiskID]://"
        // This code path is for backward compatibility, we used to export the absolute path for
        // csv files to copy.cypher files.
        return filePath;
    }

    auto path = boundFilePath + "/" + filePath;
#if defined(_WIN32)
    // TODO(Ziyi): This is a temporary workaround because our parser requires input cypher queries
    // to escape all special characters in string literal. E.g. The user input query is: 'IMPORT
    // DATABASE 'C:\\db\\uw''. The parser removes any escaped characters and this function accepts
    // the path parameter as 'C:\db\uw'. Then the ImportDatabase operator gives the file path to
    // antlr4 parser directly without escaping any special characters in the path, which causes a
    // parser exception. However, the parser exception is not thrown properly which leads to the
    // undefined behaviour.
    size_t pos = 0;
    while ((pos = path.find('\\', pos)) != std::string::npos) {
        path.replace(pos, 1, "\\\\");
        pos += 2;
    }
#endif
    return path;
}

std::unique_ptr<BoundStatement> Binder::bindImportDatabaseClause(const Statement& statement) {
    auto& importDB = statement.constCast<ImportDB>();
    auto fs = VirtualFileSystem::GetUnsafe(*clientContext);
    auto boundFilePath = fs->expandPath(clientContext, importDB.getFilePath());
    if (!fs->fileOrPathExists(boundFilePath, clientContext)) {
        throw BinderException(std::format("Directory {} does not exist.", boundFilePath));
    }
    std::string finalQueryStatements;
    finalQueryStatements +=
        getQueryFromFile(fs, boundFilePath, PortDBConstants::SCHEMA_FILE_NAME, clientContext);
    // replace the path in copy from statements with the bound path
    auto copyQuery =
        getQueryFromFile(fs, boundFilePath, PortDBConstants::COPY_FILE_NAME, clientContext);
    if (!copyQuery.empty()) {
        auto parsedStatements = Parser::parseQuery(copyQuery);
        for (auto& parsedStatement : parsedStatements) {
            DASSERT(parsedStatement->getStatementType() == StatementType::COPY_FROM);
            auto& copyFromStatement = parsedStatement->constCast<CopyFrom>();
            DASSERT(copyFromStatement.getSource()->type == ScanSourceType::FILE);
            auto filePaths =
                copyFromStatement.getSource()->constPtrCast<FileScanSource>()->filePaths;
            DASSERT(filePaths.size() == 1);
            auto fileTypeInfo = bindFileTypeInfo(filePaths);
            std::string query;
            auto copyFilePath = getCopyFilePath(boundFilePath, filePaths[0]);
            auto columnNames = getColumnNamesToCopy(copyFromStatement);
            auto parsingOptions = bindParsingOptions(copyFromStatement.getParsingOptions());
            std::unordered_map<std::string, std::string> copyFromOptions;
            if (parsingOptions.contains(CopyConstants::FROM_OPTION_NAME)) {
                DASSERT(parsingOptions.contains(CopyConstants::TO_OPTION_NAME));
                copyFromOptions[CopyConstants::FROM_OPTION_NAME] = std::format("'{}'",
                    parsingOptions.at(CopyConstants::FROM_OPTION_NAME).getValue<std::string>());
                copyFromOptions[CopyConstants::TO_OPTION_NAME] = std::format("'{}'",
                    parsingOptions.at(CopyConstants::TO_OPTION_NAME).getValue<std::string>());
                parsingOptions.erase(CopyConstants::FROM_OPTION_NAME);
                parsingOptions.erase(CopyConstants::TO_OPTION_NAME);
            }
            if (fileTypeInfo.fileType == FileType::CSV) {
                auto csvConfig = CSVReaderConfig::construct(parsingOptions);
                csvConfig.option.autoDetection = false;
                auto optionsMap = csvConfig.option.toOptionsMap(csvConfig.parallel);
                if (!copyFromOptions.empty()) {
                    optionsMap.insert(copyFromOptions.begin(), copyFromOptions.end());
                }
                query =
                    std::format("COPY `{}` {} FROM \"{}\" {};", copyFromStatement.getTableName(),
                        columnNames, copyFilePath, CSVOption::toCypher(optionsMap));
            } else {
                query =
                    std::format("COPY `{}` {} FROM \"{}\" {};", copyFromStatement.getTableName(),
                        columnNames, copyFilePath, CSVOption::toCypher(copyFromOptions));
            }
            finalQueryStatements += query;
        }
    }
    return std::make_unique<BoundImportDatabase>(boundFilePath, finalQueryStatements,
        getQueryFromFile(fs, boundFilePath, PortDBConstants::INDEX_FILE_NAME, clientContext));
}

} // namespace binder
} // namespace lbug