Skip to content

Commit 54e6181

Browse files
estebanzimanyiestebanzimanyi
committed
Initialize MEOS per worker thread
MEOS keeps the session timezone, errno, PROJ context and RNGs in thread-local storage and requires every thread that calls into it to run meos_initialize() before its first call. The extension only did this once on the load thread, so DuckDB TaskScheduler workers ran scalar, cast and aggregate bodies with a NULL session_timezone and segfaulted in pg_next_dst_boundary on the first timestamp parse. A thread-local guard now runs the per-thread init (and re-installs the process-global error handler, which meos_initialize() resets to the exit-on-error default) at the scalar exec wrapper and through a cast registration trampoline covering every cast entry point.
1 parent 05602e0 commit 54e6181

14 files changed

Lines changed: 147 additions & 80 deletions

‎src/geo/geoset.cpp‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -36,12 +36,12 @@ void SpatialSetType::RegisterTypes(ExtensionLoader &loader){
3636
}
3737

3838
void SpatialSetType::RegisterCastFunctions(ExtensionLoader &loader) {
39-
loader.RegisterCastFunction(
39+
RegisterMeosCastFunction(loader,
4040
LogicalType::VARCHAR,
4141
SpatialSetType::geomset(),
4242
SpatialSetFunctions::Text_to_geoset
4343
);
44-
loader.RegisterCastFunction(
44+
RegisterMeosCastFunction(loader,
4545
LogicalType::VARCHAR,
4646
SpatialSetType::geogset(),
4747
SpatialSetFunctions::Text_to_geoset

‎src/geo/stbox.cpp‎

Lines changed: 7 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -27,43 +27,43 @@ void StboxType::RegisterType(ExtensionLoader &loader) {
2727
}
2828

2929
void StboxType::RegisterCastFunctions(ExtensionLoader &loader) {
30-
loader.RegisterCastFunction(
30+
RegisterMeosCastFunction(loader,
3131
LogicalType::VARCHAR,
3232
STBOX(),
3333
StboxFunctions::Stbox_in_cast
3434
);
3535

36-
loader.RegisterCastFunction(
36+
RegisterMeosCastFunction(loader,
3737
STBOX(),
3838
LogicalType::VARCHAR,
3939
StboxFunctions::Stbox_out
4040
);
4141

42-
loader.RegisterCastFunction(
42+
RegisterMeosCastFunction(loader,
4343
GeoTypes::GEOMETRY(),
4444
STBOX(),
4545
StboxFunctions::Geo_to_stbox_cast
4646
);
4747

48-
loader.RegisterCastFunction(
48+
RegisterMeosCastFunction(loader,
4949
LogicalType::TIMESTAMP_TZ,
5050
STBOX(),
5151
StboxFunctions::Timestamptz_to_stbox_cast
5252
);
5353

54-
loader.RegisterCastFunction(
54+
RegisterMeosCastFunction(loader,
5555
SetTypes::tstzset(),
5656
STBOX(),
5757
StboxFunctions::Tstzset_to_stbox_cast
5858
);
5959

60-
loader.RegisterCastFunction(
60+
RegisterMeosCastFunction(loader,
6161
SpanTypes::TSTZSPAN(),
6262
STBOX(),
6363
StboxFunctions::Tstzspan_to_stbox_cast
6464
);
6565

66-
loader.RegisterCastFunction(
66+
RegisterMeosCastFunction(loader,
6767
SpansetTypes::tstzspanset(),
6868
STBOX(),
6969
StboxFunctions::Tstzspanset_to_stbox_cast

‎src/geo/tgeogpoint.cpp‎

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -43,25 +43,25 @@ void TgeogpointType::RegisterType(ExtensionLoader &loader) {
4343
}
4444

4545
void TgeogpointType::RegisterCastFunctions(ExtensionLoader &loader) {
46-
loader.RegisterCastFunction(
46+
RegisterMeosCastFunction(loader,
4747
LogicalType::VARCHAR,
4848
TGEOGPOINT(),
4949
TgeogpointFunctions::Tpoint_in
5050
);
5151

52-
loader.RegisterCastFunction(
52+
RegisterMeosCastFunction(loader,
5353
TGEOGPOINT(),
5454
LogicalType::VARCHAR,
5555
TemporalFunctions::Temporal_out
5656
);
5757

58-
loader.RegisterCastFunction(
58+
RegisterMeosCastFunction(loader,
5959
TGEOGPOINT(),
6060
StboxType::STBOX(),
6161
TgeompointFunctions::Tspatial_to_stbox_cast
6262
);
6363

64-
loader.RegisterCastFunction(
64+
RegisterMeosCastFunction(loader,
6565
TGEOGPOINT(),
6666
SpanTypes::TSTZSPAN(),
6767
TgeompointFunctions::Temporal_to_tstzspan_cast

‎src/geo/tgeogpoint_in_out.cpp‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -2,6 +2,7 @@
22
#include "geo/tgeogpoint_functions.hpp"
33
#include "duckdb/main/extension/extension_loader.hpp"
44
#include "duckdb/common/extension_type_info.hpp"
5+
#include "mobilityduck/meos_exec_serial.hpp"
56
#include <regex>
67
#include <string>
78
#include <temporal/span.hpp>
@@ -215,8 +216,8 @@ void TGeogpointType::RegisterScalarInOutFunctions(ExtensionLoader &loader){
215216

216217

217218
void TGeogpointType::RegisterCastFunctions(ExtensionLoader &loader) {
218-
loader.RegisterCastFunction( LogicalType::VARCHAR, TGeogpointType::TGEOGPOINT(), TgeogpointFunctions::StringToTgeogpoint);
219-
loader.RegisterCastFunction( TGeogpointType::TGEOGPOINT(), LogicalType::VARCHAR, TgeogpointFunctions::TgeogpointToString);
219+
RegisterMeosCastFunction(loader, LogicalType::VARCHAR, TGeogpointType::TGEOGPOINT(), TgeogpointFunctions::StringToTgeogpoint);
220+
RegisterMeosCastFunction(loader, TGeogpointType::TGEOGPOINT(), LogicalType::VARCHAR, TgeogpointFunctions::TgeogpointToString);
220221
}
221222

222223
}

‎src/geo/tgeography_in_out.cpp‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -288,8 +288,8 @@ void TGeographyTypes::RegisterScalarInOutFunctions(ExtensionLoader &loader){
288288

289289

290290
void TGeographyTypes::RegisterCastFunctions(ExtensionLoader &loader) {
291-
loader.RegisterCastFunction( LogicalType::VARCHAR, TGeographyTypes::TGEOGRAPHY(), TgeographyFunctions::StringToTgeography);
292-
loader.RegisterCastFunction( TGeographyTypes::TGEOGRAPHY(), LogicalType::VARCHAR, TgeographyFunctions::TgeographyToString);
291+
RegisterMeosCastFunction(loader, LogicalType::VARCHAR, TGeographyTypes::TGEOGRAPHY(), TgeographyFunctions::StringToTgeography);
292+
RegisterMeosCastFunction(loader, TGeographyTypes::TGEOGRAPHY(), LogicalType::VARCHAR, TgeographyFunctions::TgeographyToString);
293293
}
294294

295295
}

‎src/geo/tgeometry_in_out.cpp‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -292,8 +292,8 @@ void TGeometryTypes::RegisterScalarInOutFunctions(ExtensionLoader &loader){
292292

293293

294294
void TGeometryTypes::RegisterCastFunctions(ExtensionLoader &loader) {
295-
loader.RegisterCastFunction( LogicalType::VARCHAR, TGeometryTypes::TGEOMETRY(), TgeometryFunctions::StringToTgeometry);
296-
loader.RegisterCastFunction( TGeometryTypes::TGEOMETRY(), LogicalType::VARCHAR, TgeometryFunctions::TgeometryToString);
295+
RegisterMeosCastFunction(loader, LogicalType::VARCHAR, TGeometryTypes::TGEOMETRY(), TgeometryFunctions::StringToTgeometry);
296+
RegisterMeosCastFunction(loader, TGeometryTypes::TGEOMETRY(), LogicalType::VARCHAR, TgeometryFunctions::TgeometryToString);
297297
}
298298

299299
}

‎src/geo/tgeompoint.cpp‎

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -34,25 +34,25 @@ void TgeompointType::RegisterType(ExtensionLoader &loader) {
3434
}
3535

3636
void TgeompointType::RegisterCastFunctions(ExtensionLoader &loader) {
37-
loader.RegisterCastFunction(
37+
RegisterMeosCastFunction(loader,
3838
LogicalType::VARCHAR,
3939
TGEOMPOINT(),
4040
TgeompointFunctions::Tpoint_in
4141
);
4242

43-
loader.RegisterCastFunction(
43+
RegisterMeosCastFunction(loader,
4444
TGEOMPOINT(),
4545
LogicalType::VARCHAR,
4646
TemporalFunctions::Temporal_out
4747
);
4848

49-
loader.RegisterCastFunction(
49+
RegisterMeosCastFunction(loader,
5050
TGEOMPOINT(),
5151
StboxType::STBOX(),
5252
TgeompointFunctions::Tspatial_to_stbox_cast
5353
);
5454

55-
loader.RegisterCastFunction(
55+
RegisterMeosCastFunction(loader,
5656
TGEOMPOINT(),
5757
SpanTypes::TSTZSPAN(),
5858
TgeompointFunctions::Temporal_to_tstzspan_cast

‎src/include/mobilityduck/meos_exec_serial.hpp‎

Lines changed: 32 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2,8 +2,10 @@
22

33
#include <mutex>
44

5+
#include "duckdb/function/cast/default_casts.hpp"
56
#include "duckdb/function/scalar_function.hpp"
67
#include "duckdb/main/extension/extension_loader.hpp"
8+
#include "mobilityduck/meos_thread.hpp"
79

810
namespace duckdb {
911

@@ -26,6 +28,7 @@ inline ScalarFunction WrapScalarFunctionWithMeosExecMutex(ScalarFunction sf) {
2628
scalar_function_t orig = std::move(sf.function);
2729
sf.function = [orig = std::move(orig)](DataChunk &args, ExpressionState &state, Vector &result) {
2830
std::lock_guard<std::mutex> guard(MeosSerializedExecMutex());
31+
EnsureMeosThreadInitialized();
2932
orig(args, state, result);
3033
};
3134
return sf;
@@ -35,4 +38,33 @@ inline void RegisterSerializedScalarFunction(ExtensionLoader &loader, ScalarFunc
3538
loader.RegisterFunction(WrapScalarFunctionWithMeosExecMutex(std::move(sf)));
3639
}
3740

41+
/**
42+
* Cast functions are a separate registration path from scalar functions and
43+
* have no shared execution wrapper, yet they call MEOS just the same (e.g. the
44+
* VARCHAR -> tgeompoint parse). The original function pointer is stashed in
45+
* the bound cast data and reached through a trampoline that runs the
46+
* per-thread MEOS init before delegating. MobilityDuck cast functions do not
47+
* use cast_data themselves, so forwarding it untouched is safe.
48+
*/
49+
struct MeosCastData : BoundCastData {
50+
explicit MeosCastData(cast_function_t orig_p) : orig(orig_p) {
51+
}
52+
cast_function_t orig;
53+
unique_ptr<BoundCastData> Copy() const override {
54+
return make_uniq<MeosCastData>(orig);
55+
}
56+
};
57+
58+
inline bool MeosCastTrampoline(Vector &source, Vector &result, idx_t count, CastParameters &parameters) {
59+
EnsureMeosThreadInitialized();
60+
auto &data = parameters.cast_data->Cast<MeosCastData>();
61+
return data.orig(source, result, count, parameters);
62+
}
63+
64+
inline void RegisterMeosCastFunction(ExtensionLoader &loader, const LogicalType &source, const LogicalType &target,
65+
cast_function_t function, int64_t implicit_cast_cost = -1) {
66+
loader.RegisterCastFunction(source, target, BoundCastInfo(MeosCastTrampoline, make_uniq<MeosCastData>(function)),
67+
implicit_cast_cost);
68+
}
69+
3870
} // namespace duckdb

‎src/include/mobilityduck/meos_thread.hpp‎

Lines changed: 34 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,34 @@
1+
#pragma once
2+
3+
extern "C" {
4+
#include <meos.h>
5+
}
6+
7+
// Defined in mobilityduck_extension.cpp. Converts MEOS errors into DuckDB
8+
// exceptions instead of the process-exiting default handler.
9+
extern "C" void MobilityduckMeosErrorHandler(int errlevel, int errcode, const char *errmsg);
10+
11+
namespace duckdb {
12+
13+
// MEOS keeps the session timezone, errno, PROJ context and the RNGs in
14+
// thread-local storage; each thread that calls MEOS must initialise it
15+
// before its first call (see meos.h, "Multithreading"). DuckDB runs
16+
// scalar, cast and aggregate bodies on TaskScheduler worker threads, so a
17+
// one-shot init on the load thread leaves workers with a NULL
18+
// session_timezone and pg_next_dst_boundary segfaults on the first
19+
// timestamp parse. This runs the per-thread init exactly once per thread.
20+
//
21+
// meos_initialize() resets the process-global error handler to the
22+
// exit-on-error default, so MobilityduckMeosErrorHandler is re-installed
23+
// here; the store is an idempotent atomic write of the same pointer.
24+
inline void EnsureMeosThreadInitialized() {
25+
static thread_local const bool meos_thread_ready = []() {
26+
meos_initialize();
27+
meos_initialize_error_handler(&MobilityduckMeosErrorHandler);
28+
meos_initialize_timezone("Europe/Brussels");
29+
return true;
30+
}();
31+
(void) meos_thread_ready;
32+
}
33+
34+
} // namespace duckdb

‎src/temporal/set.cpp‎

Lines changed: 7 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -93,43 +93,43 @@ LogicalType SetTypeMapping::GetChildType(const LogicalType &type) {
9393
// Register all cast functions
9494
void SetTypes::RegisterCastFunctions(ExtensionLoader &loader) {
9595
for (const auto &set_type : SetTypes::AllTypes()) {
96-
loader.RegisterCastFunction(
96+
RegisterMeosCastFunction(loader,
9797
set_type,
9898
LogicalType::VARCHAR,
9999
SetFunctions::Set_to_text
100100
); // Blob to text
101-
loader.RegisterCastFunction(
101+
RegisterMeosCastFunction(loader,
102102
LogicalType::VARCHAR,
103103
set_type,
104104
SetFunctions::Text_to_set
105105
); // text to blob
106106

107107
auto base_type = SetTypeMapping::GetChildType(set_type);
108-
loader.RegisterCastFunction(
108+
RegisterMeosCastFunction(loader,
109109
base_type,
110110
set_type,
111111
SetFunctions::Value_to_set_cast // set from base type
112112
);
113113

114-
loader.RegisterCastFunction(
114+
RegisterMeosCastFunction(loader,
115115
SetTypes::intset(),
116116
SetTypes::floatset(),
117117
SetFunctions::Intset_to_floatset_cast // intset -> floatset
118118
);
119119

120-
loader.RegisterCastFunction(
120+
RegisterMeosCastFunction(loader,
121121
SetTypes::floatset(),
122122
SetTypes::intset(),
123123
SetFunctions::Floatset_to_intset_cast // floatset --> intset
124124
);
125125

126-
loader.RegisterCastFunction(
126+
RegisterMeosCastFunction(loader,
127127
SetTypes::dateset(),
128128
SetTypes::tstzset(),
129129
SetFunctions::Dateset_to_tstzset_cast // dateset -> tstzset
130130
);
131131

132-
loader.RegisterCastFunction(
132+
RegisterMeosCastFunction(loader,
133133
SetTypes::tstzset(),
134134
SetTypes::dateset(),
135135
SetFunctions::Tstzset_to_dateset_cast // tstz -> dateset

0 commit comments

Comments
 (0)