diff --git a/src/main/java/org/mobilitydb/spark/MeosThread.java b/src/main/java/org/mobilitydb/spark/MeosThread.java deleted file mode 100644 index 1fbe402..0000000 --- a/src/main/java/org/mobilitydb/spark/MeosThread.java +++ /dev/null @@ -1,99 +0,0 @@ -/***************************************************************************** - * - * This MobilityDB code is provided under The PostgreSQL License. - * Copyright (c) 2020-2026, Université libre de Bruxelles and MobilityDB - * contributors - * - * Permission to use, copy, modify, and distribute this software and its - * documentation for any purpose, without fee, and without a written - * agreement is hereby granted, provided that the above copyright notice and - * this paragraph and the following two paragraphs appear in all copies. - * - * IN NO EVENT SHALL UNIVERSITE LIBRE DE BRUXELLES BE LIABLE TO ANY PARTY FOR - * DIRECT, INDIRECT, SPECIAL, INCIDENTAL, OR CONSEQUENTIAL DAMAGES, INCLUDING - * LOST PROFITS, ARISING OUT OF THE USE OF THIS SOFTWARE AND ITS DOCUMENTATION, - * EVEN IF UNIVERSITE LIBRE DE BRUXELLES HAS BEEN ADVISED OF THE POSSIBILITY - * OF SUCH DAMAGE. - * - * UNIVERSITE LIBRE DE BRUXELLES SPECIFICALLY DISCLAIMS ANY WARRANTIES, - * INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY - * AND FITNESS FOR A PARTICULAR PURPOSE. THE SOFTWARE PROVIDED HEREUNDER IS ON - * AN "AS IS" BASIS, AND UNIVERSITE LIBRE DE BRUXELLES HAS NO OBLIGATIONS TO - * PROVIDE MAINTENANCE, SUPPORT, UPDATES, ENHANCEMENTS, OR MODIFICATIONS. - * - *****************************************************************************/ - -package org.mobilitydb.spark; - -import functions.GeneratedFunctions; -import functions.error_handler_fn; - -/** - * MEOS initialisation for Spark executor threads. - * - * Spark runs tasks on a pool of executor threads, and MEOS setup has two - * lifetimes: process-global state (the allocator and the error handler) is - * installed once per JVM, while thread-local state (the timezone and collation - * caches, and the PROJ, GEOS and GSL contexts) belongs to each thread. This - * class installs the process-global part once and the thread-local part once - * per native thread. - * - * The entire UDF surface is generated (GeneratedSpatioTemporalUDFs); every - * generated entry point calls {@link #ensureReady()} before its first MEOS - * call, so the executor thread running it is always initialised. There are no - * hand-registered UDFs, so this class exposes only the guard — no registration - * helpers. - */ -public final class MeosThread { - - private MeosThread() {} - - /** - * No-exit MEOS error handler. MEOS's default handler calls - * exit(EXIT_FAILURE) on an ERROR, which would tear down the whole JVM if a - * MEOS error fired inside a Spark task. This handler returns instead of - * exiting; the error still surfaces because MEOS sets meos_errno, which the - * generated wrappers check (MeosErrorHandler.checkError) and rethrow as a - * Java exception. Held as a static field so JNR keeps the native callback - * alive for the process lifetime. - */ - public static final error_handler_fn NOEXIT_ERROR_HANDLER = - (errorLevel, errorCode, errorMessage) -> { /* do not exit the JVM */ }; - - /** - * Process-global MEOS setup, installed exactly once per JVM: the allocator - * and the error handler are process-global, not thread-local. The holder's - * class initialiser runs under the JVM class-initialisation lock, so a - * thread that reaches its per-thread setup always sees a fully-installed - * no-exit handler — MEOS's exiting default is never observable to another - * thread. - */ - private static final class ProcessInit { - static { - GeneratedFunctions.meos_initialize(); - GeneratedFunctions.meos_initialize_error_handler(NOEXIT_ERROR_HANDLER); - } - /** Invoking this forces the class initialiser above to run once. */ - static void ensure() { /* side effect: class initialisation */ } - } - - /** - * Per-thread MEOS setup, run once per native thread: only the thread-local - * caches. The timezone and collation caches are thread-local and set - * explicitly per thread; the PROJ, GEOS and GSL contexts are thread-local - * too and are created lazily by MEOS on first use. Full meos_initialize() is - * NOT run per thread — it re-installs the exiting default error handler that - * every other thread relies on being the no-exit one. - */ - private static final ThreadLocal MEOS_READY = ThreadLocal.withInitial(() -> { - ProcessInit.ensure(); - GeneratedFunctions.meos_initialize_timezone("UTC"); - GeneratedFunctions.meos_initialize_collation(); - return Boolean.TRUE; - }); - - /** Ensure MEOS is initialised for the calling thread. */ - public static void ensureReady() { - MEOS_READY.get(); - } -} diff --git a/tools/codegen_spark_udfs.py b/tools/codegen_spark_udfs.py index 83f31d1..7ef30e1 100644 --- a/tools/codegen_spark_udfs.py +++ b/tools/codegen_spark_udfs.py @@ -336,7 +336,6 @@ def emit_single(name, f, vis_arity=None): if params else f' spark.udf().register("{name}", ({iface}) () -> {{'] if argnames: L.append(" if (" + " || ".join(f"{a} == null" for a in argnames) + ") return null;") - L.append(" MeosThread.ensureReady();") callargs, frees = [], [] for a, p, k in zip(argnames, params, kinds): if k[0] == "ptr": @@ -465,7 +464,6 @@ def emit_dispatch(name, cands, vis_arity=None): L = [' spark.udf().register("%s", (%s) (%s) -> {' % (name, iface, ", ".join(argnames))] if argnames: L.append(" if (" + " || ".join("%s == null" % a for a in argnames) + ") return null;") - L.append(" MeosThread.ensureReady();") # order GEO/WKT-parsing candidates LAST so the strict hex parsers get first refusal def geocount(f): return sum(1 for p in classify(f)[0] if base(p["canonical"]) == "GSERIALIZED") @@ -524,7 +522,6 @@ def emit_timearg(name, op): for k in ("timestamptz", "tstzspan", "tstzset", "tstzspanset")) return (' spark.udf().register("%s", (UDF2) (a, b) -> {\n' ' if (a == null || b == null) return null;\n' - ' MeosThread.ensureReady();\n' ' jnr.ffi.Pointer t = UdfMarshal.tFromHex(a);\n' ' if (t == null) return null;\n' ' try { return UdfMarshal.restrictTime(t, b, %s); }\n' @@ -588,7 +585,6 @@ def emit_tgeoarr(name, f, shape): iface = "UDF%d<%s, %s>" % (len(argnames), ", ".join(boxes), retbox) L = [' spark.udf().register("%s", (%s) (%s) -> {' % (name, iface, ", ".join(argnames))] L.append(" if (" + " || ".join("a%d == null" % i for i in range(nA)) + ") return null;") - L.append(" MeosThread.ensureReady();") for i in range(nA): L.append(" String[] s%d = UdfMarshal.asStrArray(a%d);" % (i, i)) 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): L = [' spark.udf().register("%s", (%s) (%s) -> {' % (name, iface, ", ".join(argnames))] if argnames: L.append(" if (" + " || ".join("%s == null" % a for a in argnames) + ") return null;") - L.append(" MeosThread.ensureReady();") callargs, frees = [], [] for a, p, k in zip(argnames, ins, kinds): if k[0] == "ptr": @@ -716,7 +711,6 @@ def emit_scalar_values(name, f, shape): import org.apache.spark.sql.api.java.*; import org.apache.spark.sql.types.DataTypes; import org.mobilitydb.spark.MeosMemory; -import org.mobilitydb.spark.MeosThread; """ # Shared marshalling helpers live in their own class: the 2300+ UDFs are partitioned @@ -735,7 +729,6 @@ def emit_scalar_values(name, f, shape): import org.apache.spark.sql.RowFactory; import functions.GeneratedFunctions; import org.mobilitydb.spark.MeosMemory; -import org.mobilitydb.spark.MeosThread; final class UdfMarshal { private UdfMarshal() {} @@ -903,7 +896,6 @@ def emit_scalar_values(name, f, shape): BiFunction tnumber, BiFunction tspatial) { if (s1 == null || s2 == null) return null; - MeosThread.ensureReady(); Pointer p1 = tFromHex(s1); if (p1 == null) return null; Pointer p2 = tFromHex(s2); @@ -1335,34 +1327,7 @@ def main(): with open(os.path.join(args.out, "GeneratedSpatioTemporalUDFs.java"), "w") as fh: fh.write(main_cls) - # ── per-thread MEOS-init invariant (build-failing) ──────────────────────────── - # MEOS keeps locale/collation, session timezone, PROJ context and RNGs in THREAD- - # LOCAL storage, so every thread (Spark executor threads run UDFs off the thread - # that called meos_initialize()) must run the per-thread init guard before its first - # MEOS call — exactly the gap that crashed MobilityDuck's table functions. Assert it - # for EVERY emitted entry point so a future emit path can't silently drop the guard - # as the bindings are regenerated (the codegen north-star). A register() lambda is - # guarded iff ensureReady() / axisBool / restrictTime (the latter two call - # ensureReady() first) appears before the first GeneratedFunctions call in its body. - unguarded = [] - for fp in glob.glob(os.path.join(args.out, "*.java")): - src = open(fp).read() - for m in re.finditer(r'udf\(\)\.register\("([^"]+)"', src): - body = src[m.start():m.start() + 1600] - gf = body.find("GeneratedFunctions.") - guard = min([x for x in (body.find("ensureReady"), body.find("axisBool"), - body.find("restrictTime")) if x >= 0] or [1 << 30]) - if gf >= 0 and guard > gf: - unguarded.append("%s (%s)" % (m.group(1), os.path.basename(fp))) - if unguarded: - print("FATAL: %d generated UDF entry point(s) reach MEOS without a per-thread " - "init guard (MeosThread.ensureReady):" % len(unguarded), file=sys.stderr) - for u in unguarded[:20]: - print(" " + u, file=sys.stderr) - sys.exit(1) - print("wrote %d group classes + UdfMarshal + GeneratedSpatioTemporalUDFs in %s" % (len(written), args.out), file=sys.stderr) - print(" per-thread MEOS-init guard : every entry point verified", file=sys.stderr) print(" JMEOS functions in catalog : %d" % total, file=sys.stderr) print(" 1:1 UDFs emitted (reached) : %d (%.0f%%)" % (cov, 100.0*cov/total), file=sys.stderr) print(" internal (excluded) : %d" % internal, file=sys.stderr)