Skip to content

Commit 9d82a4a

Browse files
Retire the hand MeosThread guard for JMEOS's generated ensureReady
JMEOS now generates the per-thread MEOS-init guard: FunctionsGenerator emits GeneratedFunctions.ensureReady() and embeds it at the top of every _meos_ wrapper, so any GeneratedFunctions call initializes the calling thread on its own. Every Spark UDF reaches MEOS only through GeneratedFunctions — the marshalling helpers UdfMarshal.*FromHex delegate to GeneratedFunctions.*_from_hexwkb, and MeosMemory.free uses Unsafe with no MEOS call — so the hand MeosThread guard and the generator's explicit MeosThread.ensureReady() emissions are redundant. Stop emitting MeosThread.ensureReady() and its import, drop the now-obsolete build-failing per-thread-init invariant (JMEOS's generator owns the guard), and delete MeosThread. The handler is JMEOS's recording MeosErrorHandler, which is non-exiting and surfaces a MEOS error as a MeosException through the wrapper's checkError().
1 parent 64954cd commit 9d82a4a

2 files changed

Lines changed: 0 additions & 134 deletions

File tree

‎src/main/java/org/mobilitydb/spark/MeosThread.java‎

Lines changed: 0 additions & 99 deletions
This file was deleted.

‎tools/codegen_spark_udfs.py‎

Lines changed: 0 additions & 35 deletions
Original file line numberDiff line numberDiff line change
@@ -336,7 +336,6 @@ def emit_single(name, f, vis_arity=None):
336336
if params else f' spark.udf().register("{name}", ({iface}) () -> {{']
337337
if argnames:
338338
L.append(" if (" + " || ".join(f"{a} == null" for a in argnames) + ") return null;")
339-
L.append(" MeosThread.ensureReady();")
340339
callargs, frees = [], []
341340
for a, p, k in zip(argnames, params, kinds):
342341
if k[0] == "ptr":
@@ -465,7 +464,6 @@ def emit_dispatch(name, cands, vis_arity=None):
465464
L = [' spark.udf().register("%s", (%s) (%s) -> {' % (name, iface, ", ".join(argnames))]
466465
if argnames:
467466
L.append(" if (" + " || ".join("%s == null" % a for a in argnames) + ") return null;")
468-
L.append(" MeosThread.ensureReady();")
469467
# order GEO/WKT-parsing candidates LAST so the strict hex parsers get first refusal
470468
def geocount(f):
471469
return sum(1 for p in classify(f)[0] if base(p["canonical"]) == "GSERIALIZED")
@@ -524,7 +522,6 @@ def emit_timearg(name, op):
524522
for k in ("timestamptz", "tstzspan", "tstzset", "tstzspanset"))
525523
return (' spark.udf().register("%s", (UDF2<String, String, String>) (a, b) -> {\n'
526524
' if (a == null || b == null) return null;\n'
527-
' MeosThread.ensureReady();\n'
528525
' jnr.ffi.Pointer t = UdfMarshal.tFromHex(a);\n'
529526
' if (t == null) return null;\n'
530527
' try { return UdfMarshal.restrictTime(t, b, %s); }\n'
@@ -588,7 +585,6 @@ def emit_tgeoarr(name, f, shape):
588585
iface = "UDF%d<%s, %s>" % (len(argnames), ", ".join(boxes), retbox)
589586
L = [' spark.udf().register("%s", (%s) (%s) -> {' % (name, iface, ", ".join(argnames))]
590587
L.append(" if (" + " || ".join("a%d == null" % i for i in range(nA)) + ") return null;")
591-
L.append(" MeosThread.ensureReady();")
592588
for i in range(nA):
593589
L.append(" String[] s%d = UdfMarshal.asStrArray(a%d);" % (i, i))
594590
L.append(" if (" + " || ".join("s%d == null" % i for i in range(nA)) + ") return null;")
@@ -680,7 +676,6 @@ def emit_scalar_values(name, f, shape):
680676
L = [' spark.udf().register("%s", (%s) (%s) -> {' % (name, iface, ", ".join(argnames))]
681677
if argnames:
682678
L.append(" if (" + " || ".join("%s == null" % a for a in argnames) + ") return null;")
683-
L.append(" MeosThread.ensureReady();")
684679
callargs, frees = [], []
685680
for a, p, k in zip(argnames, ins, kinds):
686681
if k[0] == "ptr":
@@ -716,7 +711,6 @@ def emit_scalar_values(name, f, shape):
716711
import org.apache.spark.sql.api.java.*;
717712
import org.apache.spark.sql.types.DataTypes;
718713
import org.mobilitydb.spark.MeosMemory;
719-
import org.mobilitydb.spark.MeosThread;
720714
"""
721715

722716
# Shared marshalling helpers live in their own class: the 2300+ UDFs are partitioned
@@ -735,7 +729,6 @@ def emit_scalar_values(name, f, shape):
735729
import org.apache.spark.sql.RowFactory;
736730
import functions.GeneratedFunctions;
737731
import org.mobilitydb.spark.MeosMemory;
738-
import org.mobilitydb.spark.MeosThread;
739732
740733
final class UdfMarshal {
741734
private UdfMarshal() {}
@@ -903,7 +896,6 @@ def emit_scalar_values(name, f, shape):
903896
BiFunction<Pointer, Pointer, Boolean> tnumber,
904897
BiFunction<Pointer, Pointer, Boolean> tspatial) {
905898
if (s1 == null || s2 == null) return null;
906-
MeosThread.ensureReady();
907899
Pointer p1 = tFromHex(s1);
908900
if (p1 == null) return null;
909901
Pointer p2 = tFromHex(s2);
@@ -1335,34 +1327,7 @@ def main():
13351327
with open(os.path.join(args.out, "GeneratedSpatioTemporalUDFs.java"), "w") as fh:
13361328
fh.write(main_cls)
13371329

1338-
# ── per-thread MEOS-init invariant (build-failing) ────────────────────────────
1339-
# MEOS keeps locale/collation, session timezone, PROJ context and RNGs in THREAD-
1340-
# LOCAL storage, so every thread (Spark executor threads run UDFs off the thread
1341-
# that called meos_initialize()) must run the per-thread init guard before its first
1342-
# MEOS call — exactly the gap that crashed MobilityDuck's table functions. Assert it
1343-
# for EVERY emitted entry point so a future emit path can't silently drop the guard
1344-
# as the bindings are regenerated (the codegen north-star). A register() lambda is
1345-
# guarded iff ensureReady() / axisBool / restrictTime (the latter two call
1346-
# ensureReady() first) appears before the first GeneratedFunctions call in its body.
1347-
unguarded = []
1348-
for fp in glob.glob(os.path.join(args.out, "*.java")):
1349-
src = open(fp).read()
1350-
for m in re.finditer(r'udf\(\)\.register\("([^"]+)"', src):
1351-
body = src[m.start():m.start() + 1600]
1352-
gf = body.find("GeneratedFunctions.")
1353-
guard = min([x for x in (body.find("ensureReady"), body.find("axisBool"),
1354-
body.find("restrictTime")) if x >= 0] or [1 << 30])
1355-
if gf >= 0 and guard > gf:
1356-
unguarded.append("%s (%s)" % (m.group(1), os.path.basename(fp)))
1357-
if unguarded:
1358-
print("FATAL: %d generated UDF entry point(s) reach MEOS without a per-thread "
1359-
"init guard (MeosThread.ensureReady):" % len(unguarded), file=sys.stderr)
1360-
for u in unguarded[:20]:
1361-
print(" " + u, file=sys.stderr)
1362-
sys.exit(1)
1363-
13641330
print("wrote %d group classes + UdfMarshal + GeneratedSpatioTemporalUDFs in %s" % (len(written), args.out), file=sys.stderr)
1365-
print(" per-thread MEOS-init guard : every entry point verified", file=sys.stderr)
13661331
print(" JMEOS functions in catalog : %d" % total, file=sys.stderr)
13671332
print(" 1:1 UDFs emitted (reached) : %d (%.0f%%)" % (cov, 100.0*cov/total), file=sys.stderr)
13681333
print(" internal (excluded) : %d" % internal, file=sys.stderr)

0 commit comments

Comments
 (0)