Skip to content

Commit 8f1176d

Browse files
myrrcmyrrc
committed
initial
1 parent 198cbd5 commit 8f1176d

8 files changed

Lines changed: 408 additions & 3 deletions

File tree

‎.github/workflows/ci.yml‎

Lines changed: 19 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -558,14 +558,31 @@ jobs:
558558
cargo +$NIGHTLY_TOOLCHAIN build --locked --no-default-features \
559559
--target x86_64-unknown-linux-gnu -Zbuild-std \
560560
-p vortex-ffi
561-
- name: Build C++ library (asan)
561+
- name: Build C++ tests and examples (asan)
562562
run: |
563563
mkdir -p build
564-
cmake -S lang/cpp -Bbuild -DSANITIZER=asan -DBUILD_TESTS=1 -DTARGET_TRIPLE="x86_64-unknown-linux-gnu"
564+
cmake -S lang/cpp -Bbuild -DSANITIZER=asan -DBUILD_TESTS=1 -DBUILD_EXAMPLES=1 -DTARGET_TRIPLE="x86_64-unknown-linux-gnu"
565565
cmake --build build --parallel $(nproc)
566566
- name: Run C++ tests
567567
run: |
568568
build/tests/vortex_cxx_test
569+
- name: Run C++ examples
570+
run: |
571+
cd build/examples
572+
# glob expansion in *.vortex yields an error unless you remove arrow-nanoarrow:
573+
# what(): Object store error: Generic LocalFileSystem error:
574+
# Unable to walk dir: File system loop found:
575+
# _deps/nanoarrow-src/python/subprojects/arrow-nanoarrow points to an ancestor _deps/nanoarrow-src
576+
rm -fr ../_deps/nanoarrow-src/python/subprojects/arrow-nanoarrow
577+
578+
./writer people0.vortex
579+
./writer people1.vortex
580+
./writer me.vortex
581+
582+
./scan '*.vortex'
583+
./dtype me.vortex
584+
585+
./reader
569586
570587
sqllogic-test:
571588
name: "SQL logic tests"

‎lang/cpp/CMakeLists.txt‎

Lines changed: 7 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -125,14 +125,16 @@ if(TARGET vortex_ffi_shared)
125125
endif()
126126
endif()
127127

128-
if (BUILD_TESTS)
128+
if (BUILD_TESTS OR BUILD_EXAMPLES)
129129
FetchContent_Declare(
130130
Nanoarrow
131131
GIT_REPOSITORY https://github.com/apache/arrow-nanoarrow
132132
GIT_TAG apache-arrow-nanoarrow-0.8.0
133133
)
134134
FetchContent_MakeAvailable(Nanoarrow)
135+
endif()
135136

137+
if (BUILD_TESTS)
136138
FetchContent_Declare(
137139
Catch
138140
GIT_REPOSITORY https://github.com/catchorg/Catch2.git
@@ -145,3 +147,7 @@ if (BUILD_TESTS)
145147
include(CTest)
146148
add_subdirectory(tests)
147149
endif()
150+
151+
if (BUILD_EXAMPLES)
152+
add_subdirectory(examples)
153+
endif()

‎lang/cpp/examples/CMakeLists.txt‎

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,7 @@
1+
# SPDX-License-Identifier: Apache-2.0
2+
# SPDX-FileCopyrightText: Copyright the Vortex contributors
3+
4+
foreach(name reader writer dtype scan scan_to_arrow)
5+
add_executable(${name} ${name}.cpp)
6+
target_link_libraries(${name} PRIVATE vortex_cxx_shared nanoarrow_shared)
7+
endforeach()

‎lang/cpp/examples/dtype.cpp‎

Lines changed: 107 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,107 @@
1+
// SPDX-License-Identifier: CC-BY-4.0
2+
// SPDX-FileCopyrightText: Copyright the Vortex contributors
3+
4+
#include <iostream>
5+
#include <string_view>
6+
#include <vortex/data_source.hpp>
7+
8+
using vortex::DataSource;
9+
using vortex::DataType;
10+
using vortex::DataTypeVariant;
11+
using vortex::Session;
12+
13+
static void print_dtype(const DataType &dtype);
14+
15+
static std::string_view ptype_name(vortex::PType p) {
16+
using enum vortex::PType;
17+
switch (p) {
18+
case U8:
19+
return "uint8_t";
20+
case U16:
21+
return "uint16_t";
22+
case U32:
23+
return "uint32_t";
24+
case U64:
25+
return "uint64_t";
26+
case I8:
27+
return "int8_t";
28+
case I16:
29+
return "int16_t";
30+
case I32:
31+
return "int32_t";
32+
case I64:
33+
return "int64_t";
34+
case F16:
35+
return "float16";
36+
case F32:
37+
return "float";
38+
case F64:
39+
return "double";
40+
}
41+
return "?";
42+
}
43+
44+
static void print_struct(const DataType &dtype) {
45+
std::cout << "struct(\n";
46+
for (const auto &[name, dtype] : dtype.fields()) {
47+
std::cout << " " << name << " = ";
48+
print_dtype(dtype);
49+
}
50+
std::cout << ")";
51+
}
52+
53+
static void print_dtype(const DataType &dtype) {
54+
using enum DataTypeVariant;
55+
switch (dtype.variant()) {
56+
case Null:
57+
std::cout << "null";
58+
break;
59+
case Bool:
60+
std::cout << "bool";
61+
break;
62+
case Utf8:
63+
std::cout << "utf8";
64+
break;
65+
case Binary:
66+
std::cout << "binary";
67+
break;
68+
case Extension:
69+
std::cout << "extension";
70+
break;
71+
case Primitive:
72+
std::cout << "primitive(" << ptype_name(dtype.primitive_type()) << ")";
73+
break;
74+
case Struct:
75+
print_struct(dtype);
76+
break;
77+
case List:
78+
std::cout << "list(";
79+
print_dtype(dtype.list_element());
80+
std::cout << ")";
81+
break;
82+
case FixedSizeList:
83+
std::cout << "fixed_list(";
84+
print_dtype(dtype.fixed_size_list_element());
85+
std::cout << ")";
86+
break;
87+
case Decimal:
88+
std::cout << "decimal(precision=" << static_cast<unsigned>(dtype.decimal_precision())
89+
<< ", scale=" << static_cast<int>(dtype.decimal_scale()) << ")";
90+
break;
91+
}
92+
std::cout << (dtype.nullable() ? '?' : ' ') << '\n';
93+
}
94+
95+
int main(int argc, char **argv) {
96+
if (argc != 2) {
97+
std::cerr << "Usage: dtype <file glob>\n";
98+
return 1;
99+
}
100+
101+
Session session;
102+
DataSource ds = DataSource::open(session, {argv[1]});
103+
DataType dt = ds.dtype();
104+
std::cout << "dtype: ";
105+
print_dtype(dt);
106+
return 0;
107+
}

‎lang/cpp/examples/reader.cpp‎

Lines changed: 33 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,33 @@
1+
// SPDX-License-Identifier: Apache-2.0
2+
// SPDX-FileCopyrightText: Copyright the Vortex contributors
3+
4+
#include <iostream>
5+
6+
#include <vortex/data_source.hpp>
7+
#include <vortex/writer.hpp>
8+
9+
using namespace std::string_view_literals;
10+
using namespace vortex;
11+
using namespace expr;
12+
using namespace ops; // overloaded >= for Expressions
13+
namespace fs = std::filesystem;
14+
15+
int main() {
16+
const Session session;
17+
const DataSource ds = DataSource::open(session, {"people*.vortex", "me.vortex"});
18+
Scan scan = ds.scan({.filter = col("height") >= lit<uint16_t>(50)});
19+
20+
for (Partition &partition : scan.partitions()) {
21+
for (Array &array : partition.batches()) {
22+
const Array age = array.field("age");
23+
const PrimitiveView<uint8_t> age_view = age.values<uint8_t>(session);
24+
const std::span<const uint8_t> age_values = age_view.values();
25+
for (uint8_t value : age_values) {
26+
std::cout << int(value) << " ";
27+
}
28+
}
29+
}
30+
std::cout << "\n";
31+
32+
return 0;
33+
}

‎lang/cpp/examples/scan.cpp‎

Lines changed: 116 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,116 @@
1+
// SPDX-License-Identifier: CC-BY-4.0
2+
// SPDX-FileCopyrightText: Copyright the Vortex contributors
3+
4+
#include <unistd.h>
5+
6+
#include <cstdlib>
7+
#include <iostream>
8+
#include <thread>
9+
#include <vector>
10+
11+
#include <vortex/data_source.hpp>
12+
#include <vortex/estimate.hpp>
13+
14+
using vortex::DataSource;
15+
using vortex::Estimate;
16+
using vortex::Scan;
17+
using vortex::Session;
18+
19+
static void print_estimate(const char *what, const Estimate &est) {
20+
using enum vortex::EstimateType;
21+
switch (est.type()) {
22+
case Unknown:
23+
std::cout << what << ": unknown\n";
24+
break;
25+
case Exact:
26+
std::cout << what << ": " << est.value() << '\n';
27+
break;
28+
case Inexact:
29+
std::cout << what << ": at most " << est.value() << '\n';
30+
break;
31+
}
32+
}
33+
34+
struct ScanStats {
35+
size_t partitions = 0;
36+
size_t arrays = 0;
37+
size_t rows = 0;
38+
};
39+
40+
static ScanStats worker(Scan &scan) {
41+
ScanStats stats;
42+
while (auto partition = scan.next_partition()) {
43+
++stats.partitions;
44+
while (auto array = partition->next()) {
45+
++stats.arrays;
46+
stats.rows += array->size();
47+
}
48+
}
49+
return stats;
50+
}
51+
52+
bool parse_arguments(int argc, char **argv, size_t &num_threads, std::string_view &files) {
53+
int opt = 0;
54+
while ((opt = getopt(argc, argv, "j:")) != -1) {
55+
switch (opt) {
56+
case 'j':
57+
num_threads = static_cast<size_t>(std::atoi(optarg));
58+
break;
59+
default:
60+
std::cerr << "Multi-threaded file scan\nUsage: scan [-j "
61+
"threads] <file glob>\n";
62+
return false;
63+
}
64+
}
65+
if (optind + 1 != argc) {
66+
std::cerr << "Multi-threaded file scan\nUsage: scan [-j threads] <file "
67+
"glob>\n";
68+
return false;
69+
}
70+
71+
files = argv[optind];
72+
return true;
73+
}
74+
75+
int main(int argc, char **argv) {
76+
size_t num_threads = 0;
77+
std::string_view files;
78+
if (!parse_arguments(argc, argv, num_threads, files)) {
79+
return 1;
80+
}
81+
std::cout << "Opening files: " << files << '\n';
82+
83+
const Session session;
84+
const DataSource ds = DataSource::open(session, {files});
85+
86+
print_estimate("Data source row count", ds.row_count());
87+
88+
Scan scan = ds.scan();
89+
print_estimate("Partition count", scan.partition_count());
90+
91+
if (num_threads == 0) {
92+
num_threads = scan.partition_count().value_or(1);
93+
}
94+
95+
std::cout << "Starting scan, using " << num_threads << " threads\n";
96+
std::vector<std::thread> threads;
97+
threads.reserve(num_threads);
98+
std::vector<ScanStats> results(num_threads);
99+
100+
for (size_t i = 0; i < num_threads; ++i) {
101+
threads.emplace_back([i, &scan, &results] { results[i] = worker(scan); });
102+
}
103+
for (auto &t : threads) {
104+
t.join();
105+
}
106+
107+
ScanStats total;
108+
for (const auto &r : results) {
109+
total.partitions += r.partitions;
110+
total.arrays += r.arrays;
111+
total.rows += r.rows;
112+
}
113+
std::cout << "Finished scan, processed " << total.partitions << " partitions, " << total.arrays
114+
<< " arrays, " << total.rows << " rows\n";
115+
return 0;
116+
}

‎lang/cpp/examples/scan_to_arrow.cpp‎

Lines changed: 64 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,64 @@
1+
// SPDX-License-Identifier: CC-BY-4.0
2+
// SPDX-FileCopyrightText: Copyright the Vortex contributors
3+
4+
#include <nanoarrow/nanoarrow.h>
5+
6+
typedef struct ArrowSchema FFI_ArrowSchema;
7+
typedef struct ArrowArray FFI_ArrowArray;
8+
typedef struct ArrowArrayStream FFI_ArrowArrayStream;
9+
#define USE_OWN_ARROW 1
10+
11+
#include <cstring>
12+
#include <iostream>
13+
#include <string_view>
14+
#include <vortex/data_source.hpp>
15+
16+
using vortex::ArrowStream;
17+
using vortex::DataSource;
18+
using vortex::DataType;
19+
using vortex::Scan;
20+
using vortex::Session;
21+
22+
int main(int argc, char **argv) {
23+
if (argc != 2) {
24+
std::cerr << "Scan vortex files to Arrow\nUsage: scan_to_arrow <file glob>\n";
25+
return 1;
26+
}
27+
const char *paths = argv[1];
28+
29+
Session session;
30+
DataSource ds = DataSource::open(session, {paths});
31+
Scan scan = ds.scan();
32+
33+
DataType out_dtype = ds.dtype();
34+
ArrowSchema schema = out_dtype.to_arrow();
35+
char schema_buf[10 * 1024];
36+
const int64_t schema_len = ArrowSchemaToString(&schema, schema_buf, sizeof schema_buf, 1);
37+
std::cout << "arrow schema: " << std::string_view {schema_buf, static_cast<size_t>(schema_len)} << '\n';
38+
if (schema.release != nullptr) {
39+
schema.release(&schema);
40+
}
41+
42+
ArrowError arrow_error;
43+
ArrowErrorInit(&arrow_error);
44+
45+
size_t partition_idx = 0;
46+
while (auto partition = scan.next_partition()) {
47+
ArrowStream stream = std::move(*partition).into_arrow_stream();
48+
49+
size_t rows = 0;
50+
size_t array_count = 0;
51+
ArrowArray array = {};
52+
while (ArrowArrayStreamGetNext(stream.raw(), &array, &arrow_error) == NANOARROW_OK &&
53+
array.release != nullptr) {
54+
rows += array.length;
55+
++array_count;
56+
array.release(&array);
57+
std::memset(&array, 0, sizeof(array));
58+
}
59+
std::cout << "Read partition " << partition_idx << " to Arrow, " << array_count << " arrays, " << rows
60+
<< " rows\n";
61+
++partition_idx;
62+
}
63+
return 0;
64+
}

0 commit comments

Comments
 (0)