Merge pull request #5 from Irval1337/added-bruh-format

Added columnar bruh format and some overall improvements

Author: Irval1337

CMakeLists.txt

@@ -13,6 +13,13 @@ include(cmake/deps.cmake)
 
 add_library(columnar_lib
   src/core/column.cpp
+  src/csv/csv_batch_reader.cpp
+  src/csv/csv_batch_writer.cpp
+  src/csv/csv_row_reader.cpp
+  src/csv/schema_manager.cpp
+  src/bruh/bruh_batch_reader.cpp
+  src/bruh/bruh_batch_writer.cpp
+  src/util/parse.cpp
 )
 
 target_include_directories(columnar_lib PUBLIC ${PROJECT_SOURCE_DIR}/include)

apps/converter/main.cpp

@@ -1,15 +1,90 @@
 #include <absl/flags/flag.h>
 #include <absl/flags/parse.h>
+#include <absl/flags/usage.h>
+#include <csv/csv.h>
+#include <bruh/bruh.h>
 
+#include <fstream>
 #include <iostream>
 
-// ABSL_FLAG(int, x, 1, "x");
-// ABSL_FLAG(int, y, 2, "y");
+ABSL_FLAG(std::string, mode, "", "Conversion mode: csv2bruh or bruh2csv");
+ABSL_FLAG(std::string, schema, "", "Path to .csv schema file");
+ABSL_FLAG(std::string, input, "", "Input file path");
+ABSL_FLAG(std::string, output, "", "Output file path");
+
+using namespace columnar;  // NOLINT
+
+std::ifstream OpenInput(const std::string& path) {
+    std::ifstream in(path, std::ios::binary);
+    if (!in || !in.good()) {
+        throw std::runtime_error("Cannot open file " + path);
+    }
+    return in;
+}
+
+std::ofstream OpenOutput(const std::string& path) {
+    std::ofstream out(path, std::ios::binary);
+    if (!out || !out.good()) {
+        throw std::runtime_error("Cannot open file " + path);
+    }
+    return out;
+}
+
+template <typename Reader, typename Writer>
+void Convert(Reader& reader, Writer& writer) {
+    while (auto batch = reader.ReadNext()) {
+        writer.Write(*batch);
+    }
+    writer.Flush();
+}
 
 int main(int argc, char** argv) {
+    absl::SetProgramUsageMessage(
+        "Amazing converter between CSV and BruhDB formats.\n\n"
+        "Usage:\n"
+        "  converter --mode=csv2bruh --schema=schema.csv --input=data.csv --output=data.bruhdb\n"
+        "  converter --mode=bruh2csv --input=data.bruhdb --output=data.csv [--schema=schema.csv]");
     absl::ParseCommandLine(argc, argv);
-    // int x = absl::GetFlag(FLAGS_x);
-    // int y = absl::GetFlag(FLAGS_y);
-    std::cout << "Hello" << "\n";
-    return 0;
+
+    auto mode = absl::GetFlag(FLAGS_mode);
+    auto input = absl::GetFlag(FLAGS_input);
+    auto output = absl::GetFlag(FLAGS_output);
+    auto schema_path = absl::GetFlag(FLAGS_schema);
+    if (mode.empty() || input.empty() || output.empty()) {
+        std::cerr << absl::ProgramUsageMessage();
+        return 1;
+    }
+
+    try {
+        if (mode == "csv2bruh") {
+            if (schema_path.empty()) {
+                throw std::runtime_error("Schema required");
+            }
+            auto schema = csv::SchemaManager::ReadFromFile(schema_path);
+            auto in = OpenInput(input);
+            auto out = OpenOutput(output);
+            csv::CSVBatchReader reader(in, schema, {});
+            bruh::BruhBatchWriter writer(out, schema);
+            Convert(reader, writer);
+        } else if (mode == "bruh2csv") {
+            auto in = OpenInput(input);
+            bruh::BruhBatchReader reader(in);
+            if (!schema_path.empty()) {
+                csv::SchemaManager::WriteToFile(schema_path, reader.GetSchema());
+            }
+            auto out = OpenOutput(output);
+            csv::CSVBatchWriter writer(out, {});
+            Convert(reader, writer);
+        } else {
+            throw std::runtime_error("Unknown mode");
+        }
+        std::cout << "Done.\n";
+        return 0;
+    } catch (const std::exception& e) {
+        std::cerr << "Error: " << e.what() << "\n";
+        return 1;
+    } catch (...) {
+        std::cerr << "Unknown error\n";
+        return 1;
+    }
 }

benchmarks/CMakeLists.txt

@@ -1,5 +1,6 @@
 add_executable(columnar_bench
   bench_csv.cpp
+  bench_bruh.cpp
 )
 
 target_link_libraries(columnar_bench

benchmarks/bench_bruh.cpp

@@ -0,0 +1,60 @@
+#include <benchmark/benchmark.h>
+#include <bruh/bruh.h>
+#include <core/schema.h>
+
+#include <sstream>
+
+using namespace columnar;  // NOLINT
+
+void BenchBruhBatchWriter(benchmark::State& state) {
+    core::Schema schema({core::Field("id", core::DataType::Int64),
+                         core::Field("name", core::DataType::String),
+                         core::Field("value", core::DataType::Double)});
+    core::Batch batch(schema);
+    for (int i = 0; i < state.range(0); ++i) {
+        batch.ColumnAt(0).AppendFromString(std::to_string(i));
+        batch.ColumnAt(1).AppendFromString("value" + std::to_string(i));
+        batch.ColumnAt(2).AppendFromString("0.123");
+    }
+
+    for (auto s : state) {
+        std::stringstream ss(std::ios::in | std::ios::out | std::ios::binary);
+        bruh::BruhBatchWriter writer(ss, schema);
+        writer.Write(batch);
+        writer.Flush();
+        benchmark::DoNotOptimize(ss.str());
+    }
+
+    state.SetItemsProcessed(state.iterations() * state.range(0));
+}
+BENCHMARK(BenchBruhBatchWriter)->Range(100, 10000);
+
+void BenchBruhBatchReader(benchmark::State& state) {
+    core::Schema schema({core::Field("id", core::DataType::Int64),
+                         core::Field("name", core::DataType::String),
+                         core::Field("value", core::DataType::Double)});
+    core::Batch batch(schema);
+    for (int i = 0; i < state.range(0); ++i) {
+        batch.ColumnAt(0).AppendFromString(std::to_string(i));
+        batch.ColumnAt(1).AppendFromString("value" + std::to_string(i));
+        batch.ColumnAt(2).AppendFromString("0.123");
+    }
+
+    std::stringstream ss(std::ios::in | std::ios::out | std::ios::binary);
+    {
+        bruh::BruhBatchWriter writer(ss, schema);
+        writer.Write(batch);
+        writer.Flush();
+    }
+    std::string data = ss.str();
+
+    for (auto s : state) {
+        std::istringstream in(data, std::ios::binary);
+        bruh::BruhBatchReader reader(in);
+        while (reader.ReadNext()) {
+        }
+    }
+
+    state.SetItemsProcessed(state.iterations() * state.range(0));
+}
+BENCHMARK(BenchBruhBatchReader)->Range(100, 10000);

cmake/deps.cmake

@@ -3,13 +3,15 @@ include(FetchContent)
 FetchContent_Declare(
   googletest
   URL https://github.com/google/googletest/archive/refs/tags/v1.15.2.zip
+  DOWNLOAD_EXTRACT_TIMESTAMP TRUE
 )
 set(gtest_force_shared_crt ON CACHE BOOL "" FORCE)
 FetchContent_MakeAvailable(googletest)
 
 FetchContent_Declare(
   benchmark
   URL https://github.com/google/benchmark/archive/refs/tags/v1.9.1.zip
+  DOWNLOAD_EXTRACT_TIMESTAMP TRUE
 )
 set(BENCHMARK_ENABLE_TESTING OFF CACHE BOOL "" FORCE)
 set(BENCHMARK_ENABLE_GTEST_TESTS OFF CACHE BOOL "" FORCE)
@@ -18,5 +20,7 @@ FetchContent_MakeAvailable(benchmark)
 FetchContent_Declare(
   absl
   URL https://github.com/abseil/abseil-cpp/archive/refs/tags/20240116.2.zip
+  DOWNLOAD_EXTRACT_TIMESTAMP TRUE
 )
-FetchContent_MakeAvailable(absl)
+set(ABSL_PROPAGATE_CXX_STD ON CACHE BOOL "" FORCE)
+FetchContent_MakeAvailable(absl)

include/bruh/bruh.h

@@ -0,0 +1,5 @@
+#pragma once
+
+#include <bruh/format.h>
+#include <bruh/bruh_batch_reader.h>
+#include <bruh/bruh_batch_writer.h>

include/bruh/bruh_batch_reader.h

@@ -0,0 +1,57 @@
+#pragma once
+
+#include <core/batch_reader.h>
+#include <bruh/format.h>
+#include <util/macro.h>
+
+#include <fstream>
+#include <memory>
+#include <string>
+#include <cstring>
+
+// (@Irval1337) TODO: Maybe implement file io using mmap?
+namespace columnar::bruh {
+class BruhBatchReader final : public core::BatchReader {
+public:
+    BruhBatchReader(std::istream& is) : is_(is), curr_row_group_(0) {
+        ReadMetaData();
+    }
+
+    std::optional<core::Batch> ReadNext() override {
+        if (curr_row_group_ >= metadata_.row_groups.size()) {
+            return std::nullopt;
+        }
+        return ReadRowGroup(curr_row_group_++);
+    }
+
+    core::Batch ReadRowGroup(std::size_t i);
+
+    std::size_t NumRowGroups() const {
+        return metadata_.row_groups.size();
+    }
+
+    const FileMetaData& GetMetaData() const {
+        return metadata_;
+    }
+
+    const core::Schema& GetSchema() const override {
+        return metadata_.schema;
+    }
+
+private:
+    void ReadMetaData();
+
+    // Warning: You must call these methods in the same order as below
+    void EnsureBruhFormat();
+
+    void ReadSchema(uint32_t cols_count);
+
+    void ReadRowGroupsMetadata(uint32_t cols_count);
+
+    void ReadColumn(std::unique_ptr<core::Column>& col, const core::Field& field, std::size_t n);
+
+    std::istream& is_;
+    FileMetaData metadata_;
+    std::size_t curr_row_group_;
+};
+}  // namespace columnar::bruh

include/bruh/bruh_batch_writer.h

@@ -0,0 +1,42 @@
+#pragma once
+
+#include <core/batch_writer.h>
+#include <bruh/format.h>
+#include <util/macro.h>
+
+#include <fstream>
+#include <memory>
+#include <ostream>
+#include <string>
+
+namespace columnar::bruh {
+class BruhBatchWriter final : public core::BatchWriter {
+public:
+    BruhBatchWriter(std::ostream& os, const core::Schema& schema) : os_(os), schema_(schema) {
+        metadata_.version = kCurrentVersion;
+        metadata_.schema = schema;
+        WriteMagic();
+    }
+
+    void Write(const core::Batch& batch) override;
+
+    void Flush() override;
+
+private:
+    void WriteMagic() {
+        os_.write(reinterpret_cast<const char*>(kMagicBytes), sizeof(kMagicBytes));
+    }
+
+    void WriteColumn(const core::Column& col, const core::Field& field);
+
+    void WriteFooter();
+
+    void WriteFields();
+
+    void WriteRowGroups();
+
+    std::ostream& os_;
+    core::Schema schema_;
+    FileMetaData metadata_;
+};
+}  // namespace columnar::bruh

include/bruh/format.h

@@ -0,0 +1,39 @@
+#pragma once
+
+#include <core/datatype.h>
+#include <core/schema.h>
+
+#include <cstdint>
+#include <vector>
+
+// File format was inspired by https://parquet.apache.org/docs/file-format/
+// Binary representation looks like this:
+// Magic BRUHDB bytes (8 bytes)
+// Row groups with data (variable count and size)
+// Footer (contains FileMetaData)
+// Footer size (4 bytes)
+// Magic BRUHDB bytes (8 bytes)
+// My amazing DB uses ONLY little-endian bytes ordering
+namespace columnar::bruh {
+constexpr uint8_t kMagicBytes[8] = {'B', 'R', 'U', 'H', 'D', 'B', 0x67, 0x67};
+constexpr int kCurrentVersion = 1;
+
+struct ColumnChunkMetaData {
+    uint64_t offset;
+    uint64_t byte_size;  // (@Irval1337) TODO: split this into compressed and decompressed sizes
+    uint64_t values_count;
+};
+
+struct RowGroupMetaData {
+    uint64_t byte_size;
+    uint64_t rows_count;
+    std::vector<ColumnChunkMetaData> columns;
+};
+
+struct FileMetaData {
+    int version;
+    core::Schema schema;
+    uint64_t rows_count;
+    std::vector<RowGroupMetaData> row_groups;
+};
+}  // namespace columnar::bruh