Skip to content

[FLINK-36298][table-planner] Re-establish Calcite metadata provider on foreign threads#28738

Open
MartijnVisser wants to merge 1 commit into
apache:masterfrom
MartijnVisser:FLINK-36298
Open

[FLINK-36298][table-planner] Re-establish Calcite metadata provider on foreign threads#28738
MartijnVisser wants to merge 1 commit into
apache:masterfrom
MartijnVisser:FLINK-36298

Conversation

@MartijnVisser

Copy link
Copy Markdown
Contributor

What is the purpose of the change

Fixes the intermittent NullPointerException: metadataHandlerProvider (Calcite RelMetadataQueryBase) that has recurred for years, only in flink-python (e.g. StreamPandasConversionTests).

Root cause: Calcite keeps the metadata handler provider in a thread-local (RelMetadataQueryBase.THREAD_PROVIDERS), seeded only on the thread that builds the cached, shared RelOptCluster (FlinkRelOptClusterFactory). PyFlink runs optimization (to_pandas() -> ArrowUtils.collectAsPandasDataFrame -> table.execute().collect()) on a py4j gateway thread that can differ from the cluster-building thread, where the thread-local is unset, so the first metadata-handler lookup NPEs. The nondeterministic py4j thread assignment is why it is intermittent and PyFlink-only.

The fix re-seeds the thread-local in FlinkRelMetadataQuery.instance(), the single funnel through which all metadata queries are obtained (directly, via reuseOrCreate, and via the cluster's metadata-query supplier), using the same provider FlinkRelOptClusterFactory installs.

Notes for reviewers:

  • Scope: this addresses the sequential cross-thread case (the observed failures). Concurrent optimization on multiple threads against the same shared, non-thread-safe RelMetadataQuery is a separate, pre-existing concern that this change neither introduces nor fixes.
  • The thread-local is set and not cleared on foreign (pooled py4j) threads, by parity with Calcite's own RelOptCluster.setMetadataProvider seeding; the value is the stateless FlinkDefaultRelMetadataProvider singleton chain.
  • Design choice: healing the instance() funnel is provider-agnostic and robust to any foreign-thread caller, versus seeding at the PyFlink entry point, which is more invasive and PyFlink-specific.

Likely duplicates of the same NPE: FLINK-33531, FLINK-38392.

Brief change log

  • Guard FlinkRelMetadataQuery.instance() to re-seed RelMetadataQueryBase.THREAD_PROVIDERS when unset on the current thread
  • Cross-reference comment in FlinkRelOptClusterFactory documenting the single provider both sites rely on
  • Add FlinkRelMetadataQueryThreadLocalTest, a deterministic red/green reproduction of the cross-thread NPE

Verifying this change

This change added tests and can be verified as follows:

  • FlinkRelMetadataQueryThreadLocalTest builds the cluster on the main thread and issues a metadata query on a fresh worker thread; it fails with NullPointerException: metadataHandlerProvider without the fix and passes with it
  • mvn test -Dtest=FlinkRelMd* in flink-table-planner passes (315 tests)

Does this pull request potentially affect one of the following parts:

  • Dependencies (does it add or upgrade a dependency): no
  • The public API, i.e., is any changed class annotated with @Public(Evolving): no
  • The serializers: no
  • The runtime per-record code paths (performance sensitive): no
  • Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: no
  • The S3 file system connector: no

Documentation

  • Does this pull request introduce a new feature? no
  • If yes, how is the feature documented? not applicable

Was generative AI tooling used to co-author this PR?
  • Yes (Claude Code with Claude Opus 4.8)

Generated-by: Claude Opus 4.8 (1M context)

…n foreign threads

RelMetadataQueryBase.THREAD_PROVIDERS is seeded only on the thread that
builds the cached, shared RelOptCluster. When optimization runs on a
different thread (for example a PyFlink py4j gateway thread) the
thread-local is unset and metadata handlers throw NullPointerException.
Guard FlinkRelMetadataQuery.instance() to re-seed the thread-local with
the same provider FlinkRelOptClusterFactory configures.

Generated-by: Claude Opus 4.8 (1M context)
@MartijnVisser MartijnVisser requested a review from snuyanzin July 14, 2026 12:42
@flinkbot

flinkbot commented Jul 14, 2026

Copy link
Copy Markdown
Collaborator

CI report:

Bot commands The @flinkbot bot supports the following commands:
  • @flinkbot run azure re-run the last Azure build

@MartijnVisser

Copy link
Copy Markdown
Contributor Author

@flinkbot run azure

@MartijnVisser MartijnVisser marked this pull request as ready for review July 14, 2026 18:01
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

2 participants