Repository navigation
Conversation
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## master #19107 +/- ##
============================================
+ Coverage 48.13% 48.53% +0.40%
- Complexity 625 635 +10
============================================
Files 5572 5573 +1
Lines 411555 412445 +890
Branches 54193 54373 +180
============================================
+ Hits 198084 200177 +2093
+ Misses 213471 212268 -1203 ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
JackieTien97
left a comment
There was a problem hiding this comment.
Thanks for tackling this. The UDF classloader does need to stay alive for the whole table-model query lifecycle, but I have a few concerns with the current approach (roughly in order of severity; details inline):
- Possible deadlock between the new UDF read lock and
instanceExecution.computeIfAbsent(...)while aCREATE/DROP FUNCTIONis waiting for the write lock (FragmentInstanceManagerL187). - The pin targets
activeClassLoader, not the loader that defined the table UDF class.doRegisteronly reloads classes of the model being registered. With both tree and table UDFs (including right after a DataNode restart), table UDF classes can sit on an already-closed loader, and this PR does not protect them (QueryExecutionL347). - Pin leak for non-query table statements that finish during analysis and never reach a terminal state, e.g.
LOAD ... WITH ('async'='true')(QueryExecutionL171). - The read lock now spans the entire table-model analysis, including every table insert/
LOAD/DELETE/UPDATE. OneCREATE FUNCTIONcan stall all table-model reads and writes on a DataNode behind the slowest in-flight analysis (QueryExecutionL344).
A few other paths still instantiate table UDFs without a pin (schema FIs, FI deserialization), and there are some minor points inline.
On the overall approach: pinning and locking at each entry point is easy to miss for the next caller. Schema FIs, plan deserialization, schema-region UPDATE, and tree-model analysis are still uncovered, and the global read/write lock is what introduces the stall and the deadlock above. It may be simpler and more robust to fix this inside the loader management itself, for example:
- when
doRegisterrotates the active loader, reload the registered classes of all models (and haveUDFClassLoaderManager.start()deprecate the loader it replaces), soUDFTablenever references a deprecated loader; - make
UDFClassLoaderManager#initializeUDFQueryatomic on its own: readactiveClassLoaderonce andtryAcquire()it with a CAS that fails once the loader is closed, retrying on failure. Query paths then don't need the global lock at all; - or treat the classes still referenced by
UDFTableas pins, so a loader is never closed while it still defines registered functions.
Tests: there is no UT/IT for the new lifecycle. Two tests would catch issues 2 and 3:
- a UT asserting that
queryIdToUDFClassLoaderMapis empty after an asyncLOADor a no-matchUPDATE; - an IT that runs
CREATE FUNCTIONfor a tree UDF and then a table UDF that lazily loads a helper class.
The failing CI jobs (IoTDBMQTTServiceIT hanging on Windows, IoTDBSessionSimpleIT.testWriteRestartAndDeleteDB, and the PageCacheDeletionBufferTest timeout) look unrelated at first glance; a re-run would help confirm.
| List<PipelineDriverFactory> driverFactories; | ||
| if (instance.getSessionInfo() != null | ||
| && instance.getSessionInfo().getSqlDialect() == SqlDialect.TABLE) { | ||
| UDFManagementService.getInstance().acquireReadLock(); |
There was a problem hiding this comment.
This blocks on the UDF read lock inside the instanceExecution.computeIfAbsent(...) mapping function, i.e. while holding that CHM bin. Together with the read lock that QueryExecution#analyze now holds for the entire table analysis, I think this can deadlock:
- Thread A analyzes a table-model insert with new devices. It holds the read lock (
QueryExecutionL344).TableDeviceSchemaValidator→TableDeviceSchemaFetcher#fetchMissingDeviceSchemaForDataInsertionruns a nestedFetchDevicequery with timeoutLong.MAX_VALUE, whose local schema FI goes throughexecSchemaQueryFragmentInstance→instanceExecution.computeIfAbsent(...). - A
CREATE FUNCTIONarrives, andregister()→acquireWriteLock()waits for A. - Thread T dispatches a table data FI, enters this
computeIfAbsentlambda (holding bin B), and blocks here. The non-fairReentrantReadWriteLockblocks new readers once a writer is queued. - If A's schema FI lands in bin B, A blocks on that bin.
That makes the cycle A → T → writer → A. None of lock(), synchronized, or the nested fetch time out, so it never resolves. Every later table analysis or FI planning on that DataNode (and tree UDF queries via TransformOperator) then queues behind the writer, and the ConfigNode keeps holding its udfInfo lock until its DataNode requests time out. The more table FIs arrive while the writer is queued, the more bins are held, so a collision gets more likely under load.
Could we take the read lock before entering computeIfAbsent instead of blocking inside it? It would also help not to hold the read lock across the whole analysis (see the comment on QueryExecution L344).
| UDFManagementService.getInstance().acquireReadLock(); | ||
| try { | ||
| synchronized (tableUdfClassLoaderLock) { | ||
| UDFClassLoaderManager.getInstance().initializeUDFQuery(queryId); |
There was a problem hiding this comment.
This pins activeClassLoader, which is not necessarily the loader that defined the table UDF classes stored in UDFTable. UDFManagementService#doRegister rotates the global active loader on every registration, but updateAllRegisteredClasses(model, ...) only reloads classes of the model being registered. The previous loader is deprecated and, if nothing pins it, closed right away in markAsDeprecated().
Example:
CREATE FUNCTION my_scalar(table): the active loader becomes L1, and table classes are loaded by L1.CREATE FUNCTION my_udtf(tree): the active loader becomes L2, and only tree classes are reloaded. L1 is deprecated and closed immediately.SELECT my_scalar(s1) FROM tpins L2, here and inFragmentInstanceContext#initializeTableModelUDFQuery, butmy_scalar's class still comes from the closed L1. The first lazy load of a helper class, resource, orServiceLoaderentry fails withNoClassDefFoundError/ClassNotFoundException.
The same happens after a DataNode restart when both models have UDFs. prepareUDFResources() calls doRegister once per UDF, rotating each time, and UDFClassLoaderManager.start() then replaces the active loader again. So the failure this PR targets can still happen in mixed tree/table deployments. Reloading all models' classes on rotation, or pinning clazz.getClassLoader() instead of the active loader, would close this gap.
| // preventing a premature close raced with late-scheduled FragmentInstances. | ||
| context.releaseExternalTsFileQueryResources(); | ||
| this.stop(cause); | ||
| finalizeTableUDFClassLoader(); |
There was a problem hiding this comment.
For non-query statements this listener is the only place that releases the analysis pin. releaseResource() is reached through Coordinator#cleanupQueryExecution, but Coordinator#execution only puts isQuery() executions into queryExecutionMap. So a non-query table statement that never reaches a terminal state leaks its pin:
LOAD '/x.tsfile' WITH ('async'='true')in the table model:visitLoadTsFilesetsQueryType.OTHER, and a successfuldoAsyncLoadsetsfinishQueryAfterAnalyzewithout a fail status.startInternal()then callsconstructResultForMemorySource()andtransitionToRunning()and returns, so the state staysRUNNINGforever.UPDATE t SET a = 'v' WHERE <predicate that matches no device>:QueryType.WRITE, andparseWherereturns false so the analysis is marked finished.- Any non-query statement whose
start()throws fromdoLogicalPlan/doDistributedPlan/schedule().
Each of these leaves an entry in queryIdToUDFClassLoaderMap and a permanent activeQueriesCount on the loader. After the next CREATE FUNCTION, the deprecated loader is never closed: jar handles and metaspace leak, and on Windows the jar can't be deleted. Could the pin be released at the end of start() for non-query statements, or could only QueryType.READ statements be pinned?
| return planner.analyze(context); | ||
| } | ||
| String queryId = context.getQueryId().getId(); | ||
| UDFManagementService.getInstance().acquireReadLock(); |
There was a problem hiding this comment.
This holds the UDF read lock for the whole planner.analyze(context), and TableModelPlanner analysis covers every table-model statement, including inserts, LOAD, DELETE, and UPDATE. That analysis can block for a long time:
- the insert-time
FetchDeviceusesLong.MAX_VALUEas its timeout; - schema auto-creation goes through the ConfigNode;
LOADanalyzes whole TsFiles.
Once register()/deregister() is queued for the write lock, the non-fair ReentrantReadWriteLock blocks every new reader.
So one CREATE FUNCTION issued while a slow LOAD or insert is in analysis blocks all new table-model inserts and queries on that DataNode until that analysis finishes. That covers this line and FI planning in FragmentInstanceManager. Meanwhile UDFManager#createFunction on the ConfigNode keeps holding the udfInfo lock while it waits, which blocks other CREATE/DROP FUNCTION cluster-wide. If the ConfigNode-side request times out, the DataNode may still finish the registration afterwards. Before this PR, table writes were not affected by UDF DDL at all.
Could the lock be narrowed to resolving UDF classes and taking the pin? Or could it be removed from the query path with an atomic initializeUDFQuery (see the review summary)?
| dataNodeQueryContexts[0]); | ||
| List<PipelineDriverFactory> driverFactories; | ||
| if (instance.getSessionInfo() != null | ||
| && instance.getSessionInfo().getSqlDialect() == SqlDialect.TABLE) { |
There was a problem hiding this comment.
Only the data-query path is covered. Table-model schema FIs also instantiate scalar UDFs, but they go through execSchemaQueryFragmentInstance with neither the lock nor a pin:
DataNodeTableOperatorGenerator#visitTableDeviceQueryCountbuilds aDevicePredicateFilterwithColumnTransformerBuilderfortagFuzzyPredicate, e.g.COUNT DEVICES FROM t WHERE my_udf(tag1) = 'x'.- The schema-region attribute update does the same:
SchemaRegionMemoryImpl#updateTableDeviceAttribute→constructDevicePredicateUpdater, e.g.UPDATE t SET attr = my_udf(attr).
On the DataNode that hosts the schema region, nothing pins the defining loader there, and reflect() can still race with deregister().
| && instance.getSessionInfo().getSqlDialect() == SqlDialect.TABLE) { | ||
| UDFManagementService.getInstance().acquireReadLock(); | ||
| try { | ||
| context.initializeTableModelUDFQuery(); |
There was a problem hiding this comment.
For remote FIs, user table-function code has already run before this pin is taken. DataNodeInternalRPCServiceImpl#sendFragmentInstance → FragmentInstance.deserializeFrom → TableFunctionProcessorNode.deserialize calls TableFunctionFactory.getTableFunction(name).createTableFunctionHandle() and handle.deserialize(bytes). If a CREATE FUNCTION rotates the loader between deserialization and this line, the handle class belongs to the old, possibly already closed loader, while this FI pins the new active one. The window is narrow, but it is the same kind of race this PR is trying to close.
| if (!(planner instanceof TableModelPlanner)) { | ||
| return; | ||
| } | ||
| synchronized (tableUdfClassLoaderLock) { |
There was a problem hiding this comment.
analyze() holds tableUdfClassLoaderLock for the entire planner.analyze(context), and this method needs the same monitor. Suppose the query is killed or fails while retry() is re-analyzing. The state-change listener runs on the shared fixed-size MPP-Coordinator-Executor pool (coordinator_read_executor_size: 20 by default, 2 in the Edge config). It reaches releaseResource() → here and blocks for the rest of that analysis, which can be unbounded for an insert waiting on FetchDevice. That delays state notifications for other queries.
Also, as soon as the listener gets the monitor it drops the pin, while start() keeps planning; CTE materialization and scalar subqueries can still instantiate UDFs at that point. Could the monitor guard only the initialize/finalize bookkeeping (e.g. a pinned flag) instead of the whole analysis?
| // Analyze the statement in QueryContext. Generate the analysis this query need | ||
| private IAnalysis analyze(MPPQueryContext context) { | ||
| return planner.analyze(context); | ||
| if (!(planner instanceof TableModelPlanner)) { |
There was a problem hiding this comment.
Minor/perf: this condition applies to every table-model statement. High-rate insertTablet/insertRecords, LOAD, and DELETE, which never resolve UDFs, now pay for:
- the global read lock and a
synchronizedblock; - a put/remove on
queryIdToUDFClassLoaderMap; - an increment/decrement on the single
AtomicLongof the active loader.
All writer threads contend on those cache lines. Could pinning be limited to statements that can reference functions (e.g. QueryType.READ and UPDATE), or done lazily on the first UDF resolution?
| UDFManagementService.getInstance().releaseReadLock(); | ||
| } | ||
| } else { | ||
| driverFactories = |
There was a problem hiding this comment.
Nit: the two branches duplicate the planner.plan(...) call, and the dialect check repeats the one inside FragmentInstanceContext#initializeTableModelUDFQuery(). Two options would keep the paths from drifting apart:
- a boolean flag around a single
planner.plancall; - a small helper that does "read lock → pin → run", shared with
QueryExecution#analyze.
|




Summary
Protect custom UDF classloaders by pinning the exact loader that defined the reflected UDF class, for the lifetime of an explicit query or fragment scope.
The previous implementation pinned the active loader at scope entry and broadcast every loader generation to all active queries. That could pin unrelated loaders and still miss classes restored on older generations. This change moves pinning to
UDFManagementService.reflect()and keeps the loader pinned only until the owning query or fragment scope closes.Implementation
UDFClassLoaderManager.QueryScopehandles with active-action tracking and deferred close.functionClass.getClassLoader()inside the UDF registry read lock duringreflect().LocalExecutionPlanner.plan()and transfer deserialization ownership intoFragmentInstanceContext.UDFInformationfunction type.Tests
Added regression coverage for:
reflect()pinning the defining UDF classloader while a helper class is loaded lazily;information_schema.functionsreporting an external table scalar UDF correctly.Local validation is limited to static inspection and
git diff --check; Maven tests were not run locally.