diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/calcite/FlinkRelOptClusterFactory.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/calcite/FlinkRelOptClusterFactory.scala index 1c871c7b135fd..041b0a09dffe9 100644 --- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/calcite/FlinkRelOptClusterFactory.scala +++ b/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 diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/calcite/FlinkRelOptClusterFactoryTest.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/calcite/FlinkRelOptClusterFactoryTest.java new file mode 100644 index 0000000000000..da25e7d96c4d3 --- /dev/null +++ b/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 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(); + } + } +}