HIVE-29820: Refactor and improve ObjectCache in some RecordProcessors

[abstractdog]

What changes were proposed in this pull request?

• Introduces a TrackedCache nested class in RecordProcessor that pairs an ObjectCache with the list of keys retrieved through it, replacing the parallel cache + cacheKeys and dynamicValueCache + dynamicValueCacheKeys field pattern. MapRecordProcessor, ReduceRecordProcessor, and MergeFileRecordProcessor retrieve plans through the wrapper, and releaseCache() at close time releases every tracked key.
• Small cleanup in ObjectCacheFactory.getLlapObjectCache: removes a duplicate import, and swaps the get/putIfAbsent/log dance for computeIfAbsent with Objects.requireNonNull.
• Adds a regression test that pins the plan cache to per-processor mr.ObjectCache on LLAP — the safe default. See the note below on why plan sharing across fragments is deferred.

Why are the changes needed?

The wrapper class removes the two parallel-list bookkeeping pattern that had drifted between subclasses and made "which cache owns this key" hard to see at the call sites. It's the seam a future daemon-wide plan-sharing change would need.

Plan sharing across concurrent fragments on LLAP

An earlier iteration of this PR routed the plan cache through the daemon-wide LlapObjectCache (ObjectCacheFactory.getCache(..., isPlanCache=true, llapCacheAlwaysEnabled=true)) so N concurrent fragments would share one deserialised MapWork/ReduceWork instead of each running Kryo. Verifying against TestMiniLlapLocalCliDriver -Dqfile=acid_globallimit.q — which failed deterministically with FileAlreadyExistsException: .../acidtest1/delta_0000001_0000001_0000/bucket_00000_0 and NPEs in Operator.initialize — showed this is unsafe today: operator trees in every RecordProcessor subclass carry per-fragment mutable state that initializeOp() resets (FileSinkOperator.fsp, VectorGroupByOperator.aggregator, Operator.childOperatorsArray, VectorTopNKeyOperator's per-instance filter state, ...), and sharing the operator tree across concurrent fragments races on those fields — the HIVE-14433 class of failure.

Selective sharing (Map-only, per Reduce/MergeFile's own state) also failed the same qtest 2/2 with a VectorHashKeyWrapperBase.isNull NPE inside TopNKeyFilter.compareWithBoundary — Map-side operator trees carry per-fragment state too.

Kept off until an operator-tree refactor separates the plan descriptor from per-fragment state. The trap is documented on RecordProcessor#planCache so it doesn't have to be rediscovered.

Does this PR introduce any user-facing change?

No. Internal refactor. Runtime behaviour is unchanged from pre-refactor: each processor deserialises its own plan.

How was this patch tested?

New unit test TestRecordProcessor:

• retrieveReturnsUnderlyingValue — TrackedCache.retrieve is a pure pass-through to the underlying ObjectCache.
• planAndDynamicValueCachesAreIndependent — releasing one wrapper never touches the other.
• plansAreNotSharedAcrossProcessorsOnLlap — drives the real production path with LlapProxy.setDaemon(true), constructs two RecordProcessor instances through the production constructor, and asserts each loads its own plan (loader fires twice, results are distinct). Guards against the plan cache being re-flipped to llapCacheAlwaysEnabled=true without a matching operator-tree fix.

Also verified TestMiniLlapLocalCliDriver -Dqfile=acid_globallimit.q — deterministic failure on the earlier iteration, green on this PR.

[Aggarwal-Raghav]

nit: Extra casting ? in other hunks, the Callable / MapWork is not present in retrieve().

MapWork mapWork = planCache.retrieve(MAP_PLAN_KEY, () -> Utilities.getMapWork(jconf));

[Aggarwal-Raghav]

I just did java level review, The LLAP is not my forte.

[abstractdog]

UPDATE: mass test failures prove pretty well that HIVE-14433 is still valid, we cannot cache MapWork/ReduceWork until we don't take care of separating what's specific to each instance and mutable: work objects contain the whole operator pipeline, containing paths, etc.

[sonarqubecloud]

Quality Gate passed

Issues
5 New issues
0 Accepted issues

Measures
0 Security Hotspots
0.0% Coverage on New Code
0.0% Duplication on New Code

See analysis details on SonarQube Cloud