[FLINK-36298][table] Fix NullPointerException in metadata query supplier on fresh threads

flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/calcite/FlinkRelOptClusterFactory.scala

@@ -21,7 +21,7 @@ import org.apache.flink.table.planner.hint.FlinkHintStrategies
 import org.apache.flink.table.planner.plan.metadata.{FlinkDefaultRelMetadataProvider, FlinkRelMetadataQuery}
 
 import org.apache.calcite.plan.{RelOptCluster, RelOptPlanner}
-import org.apache.calcite.rel.metadata.{DefaultRelMetadataProvider, RelMetadataQuery}
+import org.apache.calcite.rel.metadata.{DefaultRelMetadataProvider, JaninoRelMetadataProvider, RelMetadataQuery, RelMetadataQueryBase}
 import org.apache.calcite.rex.RexBuilder
 
 import java.util.function.Supplier
@@ -32,11 +32,18 @@ import java.util.function.Supplier
  */
 object FlinkRelOptClusterFactory {
 
+  private val METADATA_HANDLER_PROVIDER =
+    JaninoRelMetadataProvider.of(FlinkDefaultRelMetadataProvider.INSTANCE)
+
   def create(planner: RelOptPlanner, rexBuilder: RexBuilder): RelOptCluster = {
     val cluster = RelOptCluster.create(planner, rexBuilder)
     cluster.setMetadataProvider(FlinkDefaultRelMetadataProvider.INSTANCE)
     cluster.setMetadataQuerySupplier(new Supplier[RelMetadataQuery]() {
-      def get: FlinkRelMetadataQuery = FlinkRelMetadataQuery.instance()
+      def get: FlinkRelMetadataQuery = {
+        // Ensure THREAD_PROVIDERS is set on whichever thread invokes the supplier.
+        RelMetadataQueryBase.THREAD_PROVIDERS.set(METADATA_HANDLER_PROVIDER)
+        FlinkRelMetadataQuery.instance()
+      }
     })
     cluster.setHintStrategies(FlinkHintStrategies.createHintStrategyTable())
     cluster

flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/calcite/FlinkRelOptClusterFactoryTest.java

@@ -0,0 +1,68 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.planner.calcite;
+
+import org.apache.flink.table.planner.delegation.PlannerContext;
+import org.apache.flink.table.planner.utils.PlannerMocks;
+
+import org.apache.calcite.plan.RelOptCluster;
+import org.apache.calcite.rel.metadata.RelMetadataQuery;
+import org.apache.calcite.rel.metadata.RelMetadataQueryBase;
+import org.junit.jupiter.api.Test;
+
+import java.lang.reflect.Field;
+import java.util.concurrent.atomic.AtomicReference;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Test for {@link FlinkRelOptClusterFactory}. */
+class FlinkRelOptClusterFactoryTest {
+
+    @Test
+    void testMetadataQuerySupplierUsableOnFreshThread() throws Throwable {
+        PlannerContext plannerContext = PlannerMocks.create().getPlannerContext();
+        RelOptCluster cluster = plannerContext.createRelBuilder().getCluster();
+
+        AtomicReference<Throwable> failure = new AtomicReference<>();
+        Thread freshThread =
+                new Thread(
+                        () -> {
+                            try {
+                                // Simulate a worker thread that has never had THREAD_PROVIDERS set.
+                                RelMetadataQueryBase.THREAD_PROVIDERS.remove();
+
+                                RelMetadataQuery mq = cluster.getMetadataQuerySupplier().get();
+
+                                Field providerField =
+                                        RelMetadataQueryBase.class.getDeclaredField(
+                                                "metadataHandlerProvider");
+                                providerField.setAccessible(true);
+                                assertThat(providerField.get(mq)).isNotNull();
+                            } catch (Throwable t) {
+                                failure.set(t);
+                            }
+                        });
+        freshThread.start();
+        freshThread.join();
+
+        if (failure.get() != null) {
+            throw failure.get();
+        }
+    }
+}