Skip to content

Commit 0d80806

Browse files
rustyconoverrustyconover
committed
refactor: use generated opaque wire builders
1 parent 4b7e27d commit 0d80806

3 files changed

Lines changed: 75 additions & 79 deletions

File tree

‎src/generated/vgi_protocol_schemas.hpp‎

Lines changed: 24 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -2,7 +2,7 @@
22
// GENERATED by vgi.codegen.cpp_schemas. DO NOT EDIT BY HAND.
33
//
44
// Generator: vgi-gen-cpp-schemas v1
5-
// Content hash: b688993a6899
5+
// Content hash: c8263dfee2ad
66
//
77
// To regenerate:
88
// uv run --project ~/Development/vgi-python vgi-gen-cpp-schemas \
@@ -246,6 +246,17 @@ inline const std::shared_ptr<arrow::Schema> &AttachCatalogInfoSchema() {
246246
return schema;
247247
}
248248

249+
// Origin: ForeignKeyInfo
250+
inline const std::shared_ptr<arrow::Schema> &ForeignKeyInfoSchema() {
251+
static const auto schema = arrow::schema({
252+
arrow::field("fk_columns", arrow::list(arrow::utf8()), /*nullable=*/false),
253+
arrow::field("pk_columns", arrow::list(arrow::utf8()), /*nullable=*/false),
254+
arrow::field("referenced_table", arrow::utf8(), /*nullable=*/false),
255+
arrow::field("referenced_schema_path", arrow::list(arrow::utf8()), /*nullable=*/false),
256+
});
257+
return schema;
258+
}
259+
249260
// Origin: AggregateBindRequest
250261
inline const std::shared_ptr<arrow::Schema> &AggregateBindRequestSchema() {
251262
static const auto schema = arrow::schema({
@@ -454,6 +465,18 @@ inline const std::shared_ptr<arrow::Schema> &CatalogCreateRequestSchema() {
454465
return schema;
455466
}
456467

468+
// Origin: ClientCapabilities
469+
inline const std::shared_ptr<arrow::Schema> &ClientCapabilitiesSchema() {
470+
static const auto schema = arrow::schema({
471+
arrow::field("engine", arrow::utf8(), /*nullable=*/false),
472+
arrow::field("native_formats", arrow::list(arrow::utf8()), /*nullable=*/false),
473+
arrow::field("catalogs", arrow::list(arrow::utf8()), /*nullable=*/false),
474+
arrow::field("can_stream", arrow::boolean(), /*nullable=*/false),
475+
arrow::field("filter_encodings", arrow::list(arrow::utf8()), /*nullable=*/false),
476+
});
477+
return schema;
478+
}
479+
457480
// Origin: CopyFromContext
458481
inline const std::shared_ptr<arrow::Schema> &CopyFromContextSchema() {
459482
static const auto schema = arrow::schema({

‎src/generated/vgi_request_builders.hpp‎

Lines changed: 44 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -2,14 +2,14 @@
22
// GENERATED by vgi.codegen.cpp_request_builders. DO NOT EDIT BY HAND.
33
//
44
// Generator: vgi-gen-cpp-request-builders v1
5-
// Content hash: 8405248e0a4d
5+
// Content hash: 2c548138d83c
66
//
77
// To regenerate:
88
// uv run --project ~/Development/vgi-python vgi-gen-cpp-request-builders \
99
// > ~/Development/vgi/src/generated/vgi_request_builders.hpp
1010
// ============================================================================
1111
//
12-
// Emitted: 71 builders
12+
// Emitted: 73 builders
1313
// Skipped: 0 (Complex bucket — see vgi_rpc_types.cpp)
1414
//
1515
// ============================================================================
@@ -1351,6 +1351,48 @@ inline std::shared_ptr<arrow::RecordBatch> BuildTableFunctionStatisticsParams(
13511351
return arrow::RecordBatch::Make(schema, 1, arrays);
13521352
}
13531353

1354+
// Origin: ClientCapabilities
1355+
inline std::shared_ptr<arrow::RecordBatch> BuildClientCapabilities(
1356+
const std::string& engine,
1357+
const std::vector<std::string>& native_formats,
1358+
const std::vector<std::string>& catalogs,
1359+
bool can_stream,
1360+
const std::vector<std::string>& filter_encodings) {
1361+
const auto &schema = ClientCapabilitiesSchema();
1362+
static_assert(true, "compile-time anchor — runtime field-count assert below");
1363+
std::vector<std::shared_ptr<arrow::Array>> arrays;
1364+
arrays.reserve(5);
1365+
arrays.push_back(BuildStringScalar(engine));
1366+
arrays.push_back(BuildStringListScalar(native_formats));
1367+
arrays.push_back(BuildStringListScalar(catalogs));
1368+
arrays.push_back(BuildBoolScalar(can_stream));
1369+
arrays.push_back(BuildStringListScalar(filter_encodings));
1370+
if (arrays.size() != static_cast<size_t>(schema->num_fields())) {
1371+
throw IOException("vgi codegen drift: BuildClientCapabilities produced " + std::to_string(arrays.size()) + " arrays but schema has " + std::to_string(schema->num_fields()) + " fields");
1372+
}
1373+
return arrow::RecordBatch::Make(schema, 1, arrays);
1374+
}
1375+
1376+
// Origin: ForeignKeyInfo
1377+
inline std::shared_ptr<arrow::RecordBatch> BuildForeignKeyInfo(
1378+
const std::vector<std::string>& fk_columns,
1379+
const std::vector<std::string>& pk_columns,
1380+
const std::string& referenced_table,
1381+
const std::vector<std::string>& referenced_schema_path) {
1382+
const auto &schema = ForeignKeyInfoSchema();
1383+
static_assert(true, "compile-time anchor — runtime field-count assert below");
1384+
std::vector<std::shared_ptr<arrow::Array>> arrays;
1385+
arrays.reserve(4);
1386+
arrays.push_back(BuildStringListScalar(fk_columns));
1387+
arrays.push_back(BuildStringListScalar(pk_columns));
1388+
arrays.push_back(BuildStringScalar(referenced_table));
1389+
arrays.push_back(BuildStringListScalar(referenced_schema_path));
1390+
if (arrays.size() != static_cast<size_t>(schema->num_fields())) {
1391+
throw IOException("vgi codegen drift: BuildForeignKeyInfo produced " + std::to_string(arrays.size()) + " arrays but schema has " + std::to_string(schema->num_fields()) + " fields");
1392+
}
1393+
return arrow::RecordBatch::Make(schema, 1, arrays);
1394+
}
1395+
13541396
} // namespace generated
13551397
} // namespace vgi
13561398
} // namespace duckdb

‎src/vgi_rpc_types.cpp‎

Lines changed: 7 additions & 76 deletions
Original file line numberDiff line numberDiff line change
@@ -475,48 +475,8 @@ std::vector<uint8_t> SerializeForeignKeyToIpcBytes(const std::vector<std::string
475475
const std::vector<std::string> &pk_columns,
476476
const std::string &referenced_table,
477477
const std::string &referenced_schema) {
478-
// Build a single-row batch matching the Python ForeignKeyDef format. This
479-
// nested dataclass is not currently emitted as a standalone generated schema.
480-
auto fk_schema = arrow::schema({
481-
arrow::field("fk_columns", arrow::list(arrow::utf8())),
482-
arrow::field("pk_columns", arrow::list(arrow::utf8())),
483-
arrow::field("referenced_table", arrow::utf8()),
484-
arrow::field("referenced_schema_path", arrow::list(arrow::utf8())),
485-
});
486-
487-
// Build fk_columns array
488-
auto fk_builder = std::make_shared<arrow::ListBuilder>(arrow::default_memory_pool(),
489-
std::make_shared<arrow::StringBuilder>());
490-
auto fk_val_builder = dynamic_cast<arrow::StringBuilder *>(fk_builder->value_builder());
491-
CheckStatus(fk_builder->Append(), "fk list append");
492-
for (auto &col : fk_columns) {
493-
CheckStatus(fk_val_builder->Append(col), "fk col append");
494-
}
495-
auto fk_arr_result = fk_builder->Finish();
496-
CheckStatus(fk_arr_result.status(), "fk finish");
497-
498-
// Build pk_columns array
499-
auto pk_builder = std::make_shared<arrow::ListBuilder>(arrow::default_memory_pool(),
500-
std::make_shared<arrow::StringBuilder>());
501-
auto pk_val_builder = dynamic_cast<arrow::StringBuilder *>(pk_builder->value_builder());
502-
CheckStatus(pk_builder->Append(), "pk list append");
503-
for (auto &col : pk_columns) {
504-
CheckStatus(pk_val_builder->Append(col), "pk col append");
505-
}
506-
auto pk_arr_result = pk_builder->Finish();
507-
CheckStatus(pk_arr_result.status(), "pk finish");
508-
509-
// Build scalar string arrays
510-
arrow::StringBuilder ref_table_builder;
511-
CheckStatus(ref_table_builder.Append(referenced_table), "ref table");
512-
auto ref_table_result = ref_table_builder.Finish();
513-
CheckStatus(ref_table_result.status(), "ref table finish");
514-
515-
auto batch = arrow::RecordBatch::Make(fk_schema, 1,
516-
{fk_arr_result.ValueUnsafe(), pk_arr_result.ValueUnsafe(),
517-
ref_table_result.ValueUnsafe(),
518-
BuildStringListScalar(SingleSchemaPath(referenced_schema))});
519-
return SerializeToIpcBytes(batch);
478+
return SerializeToIpcBytes(
479+
generated::BuildForeignKeyInfo(fk_columns, pk_columns, referenced_table, SingleSchemaPath(referenced_schema)));
520480
}
521481

522482
// ============================================================================
@@ -1364,40 +1324,11 @@ TableFunctionCardinalityResult ParseTableFunctionCardinalityResult(const std::sh
13641324
//! because whether they are loadable is a per-session question this static list
13651325
//! cannot answer.
13661326
static std::vector<uint8_t> BuildClientCapabilitiesBytes() {
1367-
auto schema = arrow::schema({
1368-
arrow::field("engine", arrow::utf8(), false),
1369-
arrow::field("native_formats", arrow::list(arrow::utf8()), false),
1370-
arrow::field("catalogs", arrow::list(arrow::utf8()), false),
1371-
arrow::field("can_stream", arrow::boolean(), false),
1372-
arrow::field("filter_encodings", arrow::list(arrow::utf8()), false),
1373-
});
1374-
1375-
auto string_list = [](const std::vector<std::string> &values) {
1376-
auto value_builder = std::make_shared<arrow::StringBuilder>();
1377-
arrow::ListBuilder list_builder(arrow::default_memory_pool(), value_builder);
1378-
CheckStatus(list_builder.Append(), "open list");
1379-
for (const auto &v : values) {
1380-
CheckStatus(value_builder->Append(v), "append list value");
1381-
}
1382-
return FinishArray(list_builder, "string list");
1383-
};
1384-
1385-
std::vector<std::shared_ptr<arrow::Array>> arrays;
1386-
arrays.push_back(BuildStringScalar("duckdb"));
1387-
arrays.push_back(string_list({"parquet", "csv", "json"}));
1388-
arrays.push_back(string_list({"ducklake", "iceberg", "postgres", "mysql", "sqlite", "duckdb"}));
1389-
{
1390-
arrow::BooleanBuilder b;
1391-
// DuckDB has no streaming scan: positions ride the wire but nothing reads
1392-
// them. Saying false here is what stops a worker handing back an unbounded
1393-
// split this client could never terminate.
1394-
CheckStatus(b.Append(false), "can_stream");
1395-
arrays.push_back(FinishArray(b, "can_stream"));
1396-
}
1397-
arrays.push_back(string_list({"vgi.filters.v1"}));
1398-
1399-
auto batch = arrow::RecordBatch::Make(schema, 1, arrays);
1400-
return SerializeToIpcBytes(batch);
1327+
// DuckDB has no streaming scan: saying false prevents a worker from
1328+
// returning an unbounded split this client could never terminate.
1329+
return SerializeToIpcBytes(generated::BuildClientCapabilities(
1330+
"duckdb", {"parquet", "csv", "json"}, {"ducklake", "iceberg", "postgres", "mysql", "sqlite", "duckdb"}, false,
1331+
{"vgi.filters.v1"}));
14011332
}
14021333

14031334
std::shared_ptr<arrow::RecordBatch> BuildCatalogAttachRequest(

0 commit comments

Comments
 (0)