New C++ API examples

.github/workflows/ci.yml

@@ -558,14 +558,28 @@ jobs:
           cargo +$NIGHTLY_TOOLCHAIN build --locked --no-default-features \
             --target x86_64-unknown-linux-gnu -Zbuild-std \
             -p vortex-ffi
-      - name: Build C++ library (asan)
+      - name: Build C++ tests and examples (asan)
         run: |
           mkdir -p build
-          cmake -S lang/cpp -Bbuild -DSANITIZER=asan -DBUILD_TESTS=1 -DTARGET_TRIPLE="x86_64-unknown-linux-gnu"
+          cmake -S lang/cpp -Bbuild -DSANITIZER=asan -DBUILD_TESTS=1 -DBUILD_EXAMPLES=1 -DTARGET_TRIPLE="x86_64-unknown-linux-gnu"
           cmake --build build --parallel $(nproc)
       - name: Run C++ tests
         run: |
           build/tests/vortex_cxx_test
+      - name: Run C++ examples
+        run: |
+          cd build/examples
+          # filesystem loop detected
+          rm -fr ../_deps/nanoarrow-src/python/subprojects/arrow-nanoarrow
+
+          ./writer people0.vortex
+          ./writer people1.vortex
+          ./writer me.vortex
+
+          ./scan '*.vortex'
+          ./dtype me.vortex
+
+          ./reader
 
   sqllogic-test:
     name: "SQL logic tests"

lang/cpp/CMakeLists.txt

@@ -125,14 +125,16 @@ if(TARGET vortex_ffi_shared)
     endif()
 endif()
 
-if (BUILD_TESTS)
+if (BUILD_TESTS OR BUILD_EXAMPLES)
     FetchContent_Declare(
       Nanoarrow
       GIT_REPOSITORY https://github.com/apache/arrow-nanoarrow
       GIT_TAG        apache-arrow-nanoarrow-0.8.0
     )
     FetchContent_MakeAvailable(Nanoarrow)
+endif()
 
+if (BUILD_TESTS)
     FetchContent_Declare(
         Catch
         GIT_REPOSITORY https://github.com/catchorg/Catch2.git
@@ -145,3 +147,7 @@ if (BUILD_TESTS)
     include(CTest)
     add_subdirectory(tests)
 endif()
+
+if (BUILD_EXAMPLES)
+    add_subdirectory(examples)
+endif()

lang/cpp/examples/CMakeLists.txt

@@ -0,0 +1,7 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright the Vortex contributors
+
+foreach(name reader writer dtype scan scan_to_arrow)
+    add_executable(${name} ${name}.cpp)
+    target_link_libraries(${name} PRIVATE vortex_cxx_shared nanoarrow_shared)
+endforeach()

lang/cpp/examples/dtype.cpp

@@ -0,0 +1,107 @@
+// SPDX-License-Identifier: CC-BY-4.0
+// SPDX-FileCopyrightText: Copyright the Vortex contributors
+
+#include <iostream>
+#include <string_view>
+#include <vortex/data_source.hpp>
+
+using vortex::DataSource;
+using vortex::DataType;
+using vortex::DataTypeVariant;
+using vortex::Session;
+
+static void print_dtype(const DataType &dtype);
+
+static std::string_view ptype_name(vortex::PType p) {
+    using enum vortex::PType;
+    switch (p) {
+    case U8:
+        return "uint8_t";
+    case U16:
+        return "uint16_t";
+    case U32:
+        return "uint32_t";
+    case U64:
+        return "uint64_t";
+    case I8:
+        return "int8_t";
+    case I16:
+        return "int16_t";
+    case I32:
+        return "int32_t";
+    case I64:
+        return "int64_t";
+    case F16:
+        return "float16";
+    case F32:
+        return "float";
+    case F64:
+        return "double";
+    }
+    return "?";
+}
+
+static void print_struct(const DataType &dtype) {
+    std::cout << "struct(\n";
+    for (const auto &[name, dtype] : dtype.fields()) {
+        std::cout << "    " << name << " = ";
+        print_dtype(dtype);
+    }
+    std::cout << ")";
+}
+
+static void print_dtype(const DataType &dtype) {
+    using enum DataTypeVariant;
+    switch (dtype.variant()) {
+    case Null:
+        std::cout << "null";
+        break;
+    case Bool:
+        std::cout << "bool";
+        break;
+    case Utf8:
+        std::cout << "utf8";
+        break;
+    case Binary:
+        std::cout << "binary";
+        break;
+    case Extension:
+        std::cout << "extension";
+        break;
+    case Primitive:
+        std::cout << "primitive(" << ptype_name(dtype.primitive_type()) << ")";
+        break;
+    case Struct:
+        print_struct(dtype);
+        break;
+    case List:
+        std::cout << "list(";
+        print_dtype(dtype.list_element());
+        std::cout << ")";
+        break;
+    case FixedSizeList:
+        std::cout << "fixed_list(";
+        print_dtype(dtype.fixed_size_list_element());
+        std::cout << ")";
+        break;
+    case Decimal:
+        std::cout << "decimal(precision=" << static_cast<unsigned>(dtype.decimal_precision())
+                  << ", scale=" << static_cast<int>(dtype.decimal_scale()) << ")";
+        break;
+    }
+    std::cout << (dtype.nullable() ? '?' : ' ') << '\n';
+}
+
+int main(int argc, char **argv) {
+    if (argc != 2) {
+        std::cerr << "Usage: dtype <file glob>\n";
+        return 1;
+    }
+
+    Session session;
+    DataSource ds = DataSource::open(session, {argv[1]});
+    DataType dt = ds.dtype();
+    std::cout << "dtype: ";
+    print_dtype(dt);
+    return 0;
+}

lang/cpp/examples/reader.cpp

@@ -0,0 +1,33 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: Copyright the Vortex contributors
+
+#include <iostream>
+
+#include <vortex/data_source.hpp>
+#include <vortex/writer.hpp>
+
+using namespace std::string_view_literals;
+using namespace vortex;
+using namespace expr;
+using namespace ops; // overloaded >= for Expressions
+namespace fs = std::filesystem;
+
+int main() {
+    const Session session;
+    const DataSource ds = DataSource::open(session, {"people*.vortex", "me.vortex"});
+    Scan scan = ds.scan({.filter = col("height") >= lit<uint16_t>(50)});
+
+    for (Partition &partition : scan.partitions()) {
+        for (Array &array : partition.batches()) {
+            const Array age = array.field("age");
+            const PrimitiveView<uint8_t> age_view = age.values<uint8_t>(session);
+            const std::span<const uint8_t> age_values = age_view.values();
+            for (uint8_t value : age_values) {
+                std::cout << int(value) << " ";
+            }
+        }
+    }
+    std::cout << "\n";
+
+    return 0;
+}

lang/cpp/examples/scan.cpp

@@ -0,0 +1,106 @@
+// SPDX-License-Identifier: CC-BY-4.0
+// SPDX-FileCopyrightText: Copyright the Vortex contributors
+
+#include <unistd.h>
+
+#include <cstdlib>
+#include <iostream>
+#include <thread>
+#include <vector>
+
+#include <vortex/data_source.hpp>
+#include <vortex/estimate.hpp>
+
+using vortex::DataSource;
+using vortex::Estimate;
+using vortex::Scan;
+using vortex::Session;
+
+static void print_estimate(const char *what, const Estimate &est) {
+    using enum vortex::EstimateType;
+    switch (est.type()) {
+    case Unknown:
+        std::cout << what << ": unknown\n";
+        break;
+    case Exact:
+        std::cout << what << ": " << est.value() << '\n';
+        break;
+    case Inexact:
+        std::cout << what << ": at most " << est.value() << '\n';
+        break;
+    }
+}
+
+struct ScanStats {
+    size_t partitions = 0;
+    size_t arrays = 0;
+    size_t rows = 0;
+};
+
+static ScanStats worker(Scan &scan) {
+    ScanStats stats;
+    while (auto partition = scan.next_partition()) {
+        ++stats.partitions;
+        while (auto array = partition->next()) {
+            ++stats.arrays;
+            stats.rows += array->size();
+        }
+    }
+    return stats;
+}
+
+int main(int argc, char **argv) {
+    size_t num_threads = 0;
+    int opt = 0;
+    while ((opt = getopt(argc, argv, "j:")) != -1) {
+        switch (opt) {
+        case 'j':
+            num_threads = static_cast<size_t>(std::atoi(optarg));
+            break;
+        default:
+            std::cerr << "Multi-threaded file scan\nUsage: scan [-j "
+                         "threads] <file glob>\n";
+            return 1;
+        }
+    }
+    if (optind + 1 != argc) {
+        std::cerr << "Multi-threaded file scan\nUsage: scan [-j threads] <file "
+                     "glob>\n";
+        return 1;
+    }
+
+    const Session session;
+    std::cout << "Opening files: " << argv[optind] << '\n';
+    const DataSource ds = DataSource::open(session, {argv[optind]});
+
+    print_estimate("Data source row count", ds.row_count());
+
+    Scan scan = ds.scan();
+    print_estimate("Partition count", scan.partition_count());
+
+    if (num_threads == 0) {
+        num_threads = scan.partition_count().value_or(1);
+    }
+
+    std::cout << "Starting scan, using " << num_threads << " threads\n";
+    std::vector<std::thread> threads;
+    threads.reserve(num_threads);
+    std::vector<ScanStats> results(num_threads);
+
+    for (size_t i = 0; i < num_threads; ++i) {
+        threads.emplace_back([i, &scan, &results] { results[i] = worker(scan); });
+    }
+    for (auto &t : threads) {
+        t.join();
+    }
+
+    ScanStats total;
+    for (const auto &r : results) {
+        total.partitions += r.partitions;
+        total.arrays += r.arrays;
+        total.rows += r.rows;
+    }
+    std::cout << "Finished scan, processed " << total.partitions << " partitions, " << total.arrays
+              << " arrays, " << total.rows << " rows\n";
+    return 0;
+}

lang/cpp/examples/scan_to_arrow.cpp

@@ -0,0 +1,65 @@
+// SPDX-License-Identifier: CC-BY-4.0
+// SPDX-FileCopyrightText: Copyright the Vortex contributors
+
+#include <nanoarrow/nanoarrow.h>
+
+typedef struct ArrowSchema FFI_ArrowSchema;
+typedef struct ArrowArray FFI_ArrowArray;
+typedef struct ArrowArrayStream FFI_ArrowArrayStream;
+#define USE_OWN_ARROW 1
+
+#include <cstring>
+#include <iostream>
+#include <string_view>
+#include <vortex/data_source.hpp>
+
+using vortex::ArrowStream;
+using vortex::DataSource;
+using vortex::DataType;
+using vortex::Scan;
+using vortex::Session;
+using vortex::VortexException;
+
+int main(int argc, char **argv) {
+    if (argc != 2) {
+        std::cerr << "Scan vortex files to Arrow\nUsage: scan_to_arrow <file glob>\n";
+        return 1;
+    }
+    const char *paths = argv[1];
+
+    Session session;
+    DataSource ds = DataSource::open(session, {paths});
+    Scan scan = ds.scan();
+
+    DataType out_dtype = ds.dtype();
+    ArrowSchema schema = out_dtype.to_arrow();
+    char schema_buf[10 * 1024];
+    const int64_t schema_len = ArrowSchemaToString(&schema, schema_buf, sizeof schema_buf, 1);
+    std::cout << "arrow schema: " << std::string_view {schema_buf, static_cast<size_t>(schema_len)} << '\n';
+    if (schema.release != nullptr) {
+        schema.release(&schema);
+    }
+
+    ArrowError arrow_error;
+    ArrowErrorInit(&arrow_error);
+
+    size_t partition_idx = 0;
+    while (auto partition = scan.next_partition()) {
+        ArrowStream stream = std::move(*partition).into_arrow_stream();
+
+        size_t rows = 0;
+        size_t array_count = 0;
+        ArrowArray array = {};
+        while (ArrowArrayStreamGetNext(stream.raw(), &array, &arrow_error) == NANOARROW_OK &&
+               array.release != nullptr) {
+            rows += array.length;
+            ++array_count;
+            array.release(&array);
+            std::memset(&array, 0, sizeof(array));
+        }
+        std::cout << "Read partition " << partition_idx << " to Arrow, " << array_count << " arrays, " << rows
+                  << " rows\n";
+        ++partition_idx;
+    }
+    return 0;
+}