| 1321 | } |
| 1322 | |
| 1323 | bool sourceAppendArrowStream( |
| 1324 | void* ctx, TopicHandle topic, struct ArrowArrayStream* stream, PJ_string_view_t timestamp_column, |
| 1325 | PJ_error_t* out_error) noexcept { |
| 1326 | return guardHostCallback(out_error, [&] { |
| 1327 | auto* impl = static_cast<DatastoreSourceWriteHostState*>(ctx); |
| 1328 | auto core = impl->core.load(std::memory_order_acquire); |
| 1329 | if (!core->appendArrowStream(topic, stream, timestamp_column)) { |
| 1330 | // Failure: plugin retains ownership of the stream; we do NOT release. |
| 1331 | propagateError(out_error, core->lastError()); |
| 1332 | return false; |
| 1333 | } |
| 1334 | // Success: host now owns the stream — release it. |
| 1335 | if (stream != nullptr && stream->release != nullptr) { |
| 1336 | stream->release(stream); |
| 1337 | } |
| 1338 | return true; |
| 1339 | }); |
| 1340 | } |
| 1341 | |
| 1342 | bool parserEnsureField( |
| 1343 | void* ctx, PJ_string_view_t field_name, PJ_primitive_type_t type, FieldHandle* out_field, |
nothing calls this directly
no test coverage detected