[
https://issues.apache.org/jira/browse/CALCITE-4845?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Stamatis Zampetakis updated CALCITE-4845:
-----------------------------------------
Summary: Exceptions in RelMetadataQuery when executing HepPlanner
concurrently (was: Some exceptions in RelMetadataQuery, when executing hep
planner concurrently.)
> Exceptions in RelMetadataQuery when executing HepPlanner concurrently
> ---------------------------------------------------------------------
>
> Key: CALCITE-4845
> URL: https://issues.apache.org/jira/browse/CALCITE-4845
> Project: Calcite
> Issue Type: Bug
> Components: core
> Reporter: Xurenhe
> Priority: Major
>
> In my work, I meet some problems when executing hep planner.
> I will use a simple example to illustrate, as follow:
>
> {code:java}
> //代码占位符
> @Test void testExecPlanConcurrently() {
> final RelHint noHashJoinHint = RelHint.builder("no_hash_join").build();
> final RelBuilder builder = RelBuilder.create(config().build());
> builder.getCluster().setHintStrategies(
> HintStrategyTable.builder()
> .hintStrategy("no_hash_join", HintPredicates.JOIN)
> .build());
> final RelNode original =
> builder.scan("EMP")
> .scan("DEPT")
> .join(JoinRelType.INNER,
> builder.equals(
> builder.field(2, 0, "DEPTNO"),
> builder.field(2, 1, "DEPTNO")))
> .hints(noHashJoinHint)
> .project(
> builder.field("ENAME"),
> builder.field("DNAME"))
> .filter(builder.call(SqlStdOperatorTable.EQUALS,
> builder.field("ENAME"),
> builder.literal("hello")))
> .build();
> original.getCluster().getMetadataQuery();
> final List<RelNode> relNodes = generateMoreRels(original);
> relNodes.parallelStream().map(rel -> {
> final HepProgram hepProgram = new HepProgramBuilder()
> .addRuleInstance(CoreRules.FILTER_PROJECT_TRANSPOSE)
> .addRuleInstance(CoreRules.FILTER_INTO_JOIN)
> .addRuleInstance(CoreRules.PROJECT_JOIN_TRANSPOSE)
> .addRuleInstance(CoreRules.PROJECT_MERGE)
> .addRuleInstance(CoreRules.PROJECT_TO_CALC)
> .addRuleInstance(CoreRules.FILTER_TO_CALC)
> .addRuleInstance(CoreRules.CALC_MERGE)
> .build();
> final HepPlanner hepPlanner = new HepPlanner(hepProgram);
> hepPlanner.setRoot(rel);
> final RelNode relNode = hepPlanner.findBestExp();
> return relNode;
> }).collect(Collectors.toList());
> }
> private List<RelNode> generateMoreRels(RelNode origin) {
> final List<RelNode> relNodes = new ArrayList<>();
> for (int i = 0; i < 500; i++) {
> relNodes.add( new RelVisitor() {
> @Override
> public void visit(RelNode node, int ordinal,
> @Nullable RelNode parent) {
> super.visit(node, ordinal, parent);
> final RelNode newRel;
> if (node instanceof TableScan) {
> newRel = LogicalTableScan.create(node.getCluster(),
> ((TableScan)node).getTable(), ImmutableList.of());
> } else {
> newRel = node.copy(node.getTraitSet(), node.getInputs());
> }
> if (parent == null) {
> replaceRoot(newRel);
> } else {
> parent.replaceInput(ordinal, newRel);
> }
> }
> }.go(origin));
> }
> return relNodes;
> }
> {code}
> But, it will throw exception:
>
>
> {code:java}
> // error stack
> java.lang.NullPointerExceptionjava.lang.NullPointerException at
> sun.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method) at
> sun.reflect.NativeConstructorAccessorImpl.newInstance(NativeConstructorAccessorImpl.java:62)
> at
> sun.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45)
> at java.lang.reflect.Constructor.newInstance(Constructor.java:423) at
> java.util.concurrent.ForkJoinTask.getThrowableException(ForkJoinTask.java:598)
> at java.util.concurrent.ForkJoinTask.reportException(ForkJoinTask.java:677)
> at java.util.concurrent.ForkJoinTask.invoke(ForkJoinTask.java:735) at
> java.util.stream.ReduceOps$ReduceOp.evaluateParallel(ReduceOps.java:714) at
> java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:233) at
> java.util.stream.ReferencePipeline.collect(ReferencePipeline.java:499) at
> org.apache.calcite.test.RelConcurrentTest.testExecPlanConcurrently(RelConcurrentTest.java:108)
> at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) at
> sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
> at
> sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
> at java.lang.reflect.Method.invoke(Method.java:498) at
> org.junit.platform.commons.util.ReflectionUtils.invokeMethod(ReflectionUtils.java:725)
> at
> org.junit.jupiter.engine.execution.MethodInvocation.proceed(MethodInvocation.java:60)
> at
> org.junit.jupiter.engine.execution.InvocationInterceptorChain$ValidatingInvocation.proceed(InvocationInterceptorChain.java:131)
> at
> org.junit.jupiter.engine.extension.TimeoutInvocation.proceed(TimeoutInvocation.java:46)
> at
> org.junit.jupiter.engine.extension.TimeoutExtension.intercept(TimeoutExtension.java:149)
> at
> org.junit.jupiter.engine.extension.TimeoutExtension.interceptTestableMethod(TimeoutExtension.java:140)
> at
> org.junit.jupiter.engine.extension.TimeoutExtension.interceptTestMethod(TimeoutExtension.java:84)
> at
> org.junit.jupiter.engine.execution.ExecutableInvoker$ReflectiveInterceptorCall.lambda$ofVoidMethod$0(ExecutableInvoker.java:115)
> at
> org.junit.jupiter.engine.execution.ExecutableInvoker.lambda$invoke$0(ExecutableInvoker.java:105)
> at
> org.junit.jupiter.engine.execution.InvocationInterceptorChain$InterceptedInvocation.proceed(InvocationInterceptorChain.java:106)
> at
> org.junit.jupiter.engine.execution.InvocationInterceptorChain.proceed(InvocationInterceptorChain.java:64)
> at
> org.junit.jupiter.engine.execution.InvocationInterceptorChain.chainAndInvoke(InvocationInterceptorChain.java:45)
> at
> org.junit.jupiter.engine.execution.InvocationInterceptorChain.invoke(InvocationInterceptorChain.java:37)
> at
> org.junit.jupiter.engine.execution.ExecutableInvoker.invoke(ExecutableInvoker.java:104)
> at
> org.junit.jupiter.engine.execution.ExecutableInvoker.invoke(ExecutableInvoker.java:98)
> at
> org.junit.jupiter.engine.descriptor.TestMethodTestDescriptor.lambda$invokeTestMethod$7(TestMethodTestDescriptor.java:214)
> at
> org.junit.platform.engine.support.hierarchical.ThrowableCollector.execute(ThrowableCollector.java:73)
> at
> org.junit.jupiter.engine.descriptor.TestMethodTestDescriptor.invokeTestMethod(TestMethodTestDescriptor.java:210)
> at
> org.junit.jupiter.engine.descriptor.TestMethodTestDescriptor.execute(TestMethodTestDescriptor.java:135)
> at
> org.junit.jupiter.engine.descriptor.TestMethodTestDescriptor.execute(TestMethodTestDescriptor.java:66)
> at
> org.junit.platform.engine.support.hierarchical.NodeTestTask.lambda$executeRecursively$6(NodeTestTask.java:151)
> at
> org.junit.platform.engine.support.hierarchical.ThrowableCollector.execute(ThrowableCollector.java:73)
> at
> org.junit.platform.engine.support.hierarchical.NodeTestTask.lambda$executeRecursively$8(NodeTestTask.java:141)
> at org.junit.platform.engine.support.hierarchical.Node.around(Node.java:137)
> at
> org.junit.platform.engine.support.hierarchical.NodeTestTask.lambda$executeRecursively$9(NodeTestTask.java:139)
> at
> org.junit.platform.engine.support.hierarchical.ThrowableCollector.execute(ThrowableCollector.java:73)
> at
> org.junit.platform.engine.support.hierarchical.NodeTestTask.executeRecursively(NodeTestTask.java:138)
> at
> org.junit.platform.engine.support.hierarchical.NodeTestTask.execute(NodeTestTask.java:95)
> at
> org.junit.platform.engine.support.hierarchical.ForkJoinPoolHierarchicalTestExecutorService$ExclusiveTask.compute(ForkJoinPoolHierarchicalTestExecutorService.java:185)
> at
> org.junit.platform.engine.support.hierarchical.ForkJoinPoolHierarchicalTestExecutorService.invokeAll(ForkJoinPoolHierarchicalTestExecutorService.java:129)
> at
> org.junit.platform.engine.support.hierarchical.NodeTestTask.lambda$executeRecursively$6(NodeTestTask.java:155)
> at
> org.junit.platform.engine.support.hierarchical.ThrowableCollector.execute(ThrowableCollector.java:73)
> at
> org.junit.platform.engine.support.hierarchical.NodeTestTask.lambda$executeRecursively$8(NodeTestTask.java:141)
> at org.junit.platform.engine.support.hierarchical.Node.around(Node.java:137)
> at
> org.junit.platform.engine.support.hierarchical.NodeTestTask.lambda$executeRecursively$9(NodeTestTask.java:139)
> at
> org.junit.platform.engine.support.hierarchical.ThrowableCollector.execute(ThrowableCollector.java:73)
> at
> org.junit.platform.engine.support.hierarchical.NodeTestTask.executeRecursively(NodeTestTask.java:138)
> at
> org.junit.platform.engine.support.hierarchical.NodeTestTask.execute(NodeTestTask.java:95)
> at
> org.junit.platform.engine.support.hierarchical.ForkJoinPoolHierarchicalTestExecutorService$ExclusiveTask.compute(ForkJoinPoolHierarchicalTestExecutorService.java:185)
> at
> org.junit.platform.engine.support.hierarchical.ForkJoinPoolHierarchicalTestExecutorService.invokeAll(ForkJoinPoolHierarchicalTestExecutorService.java:129)
> at
> org.junit.platform.engine.support.hierarchical.NodeTestTask.lambda$executeRecursively$6(NodeTestTask.java:155)
> at
> org.junit.platform.engine.support.hierarchical.ThrowableCollector.execute(ThrowableCollector.java:73)
> at
> org.junit.platform.engine.support.hierarchical.NodeTestTask.lambda$executeRecursively$8(NodeTestTask.java:141)
> at org.junit.platform.engine.support.hierarchical.Node.around(Node.java:137)
> at
> org.junit.platform.engine.support.hierarchical.NodeTestTask.lambda$executeRecursively$9(NodeTestTask.java:139)
> at
> org.junit.platform.engine.support.hierarchical.ThrowableCollector.execute(ThrowableCollector.java:73)
> at
> org.junit.platform.engine.support.hierarchical.NodeTestTask.executeRecursively(NodeTestTask.java:138)
> at
> org.junit.platform.engine.support.hierarchical.NodeTestTask.execute(NodeTestTask.java:95)
> at
> org.junit.platform.engine.support.hierarchical.ForkJoinPoolHierarchicalTestExecutorService$ExclusiveTask.compute(ForkJoinPoolHierarchicalTestExecutorService.java:185)
> at java.util.concurrent.RecursiveAction.exec(RecursiveAction.java:189) at
> java.util.concurrent.ForkJoinTask.doExec(ForkJoinTask.java:289) at
> java.util.concurrent.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1056)
> at java.util.concurrent.ForkJoinPool.runWorker(ForkJoinPool.java:1692) at
> java.util.concurrent.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:157)Caused
> by: java.lang.NullPointerException: metadataProvider at
> java.util.Objects.requireNonNull(Objects.java:228) at
> org.apache.calcite.rel.metadata.RelMetadataQuery.<init>(RelMetadataQuery.java:150)
> at
> org.apache.calcite.rel.metadata.RelMetadataQuery.<init>(RelMetadataQuery.java:114)
> at
> org.apache.calcite.rel.metadata.RelMetadataQuery.instance(RelMetadataQuery.java:184)
> at
> org.apache.calcite.plan.RelOptCluster.getMetadataQuery(RelOptCluster.java:198)
> at
> org.apache.calcite.rel.logical.LogicalFilter.create(LogicalFilter.java:108)
> at
> org.apache.calcite.rel.core.RelFactories$FilterFactoryImpl.createFilter(RelFactories.java:345)
> at org.apache.calcite.tools.RelBuilder.filter(RelBuilder.java:1439) at
> org.apache.calcite.tools.RelBuilder.filter(RelBuilder.java:1399) at
> org.apache.calcite.rel.rules.FilterJoinRule.perform(FilterJoinRule.java:177)
> at
> org.apache.calcite.rel.rules.FilterJoinRule$FilterIntoJoinRule.onMatch(FilterJoinRule.java:371)
> at
> org.apache.calcite.plan.AbstractRelOptPlanner.fireRule(AbstractRelOptPlanner.java:343)
> at org.apache.calcite.plan.hep.HepPlanner.applyRule(HepPlanner.java:565) at
> org.apache.calcite.plan.hep.HepPlanner.applyRules(HepPlanner.java:428) at
> org.apache.calcite.plan.hep.HepPlanner.executeInstruction(HepPlanner.java:251)
> at
> org.apache.calcite.plan.hep.HepInstruction$RuleInstance.execute(HepInstruction.java:130)
> at
> org.apache.calcite.plan.hep.HepPlanner.executeProgram(HepPlanner.java:208) at
> org.apache.calcite.plan.hep.HepPlanner.findBestExp(HepPlanner.java:195) at
> org.apache.calcite.test.RelConcurrentTest.lambda$testExecPlanConcurrently$0(RelConcurrentTest.java:106)
> at java.util.stream.ReferencePipeline$3$1.accept(ReferencePipeline.java:193)
> at
> java.util.ArrayList$ArrayListSpliterator.forEachRemaining(ArrayList.java:1382)
> at java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:482) at
> java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:472)
> at java.util.stream.ReduceOps$ReduceTask.doLeaf(ReduceOps.java:747) at
> java.util.stream.ReduceOps$ReduceTask.doLeaf(ReduceOps.java:721) at
> java.util.stream.AbstractTask.compute(AbstractTask.java:316) at
> java.util.concurrent.CountedCompleter.exec(CountedCompleter.java:731) ... 4
> more{code}
> ----
> I I have a preliminary suspicion that
> `org.apache.calcite.plan.hep.HepRuleCall#transformTo` doing
> `rel0.getCluster().invalidateMetadataQuery();` may be not thread-safe, when
> multiple thread share the same RelOptCluster.
> For example:
> thread1: RelOptCluster is building a RMQ, because current RMQ is null.
> thread2: RelOptCluster is setting RMQ to null.
> I search
> related[CALCITE-1812|https://issues.apache.org/jira/browse/CALCITE-1812]
> Forgiving my stupidity, I don't quite understand why RelOptRuleCall need to
> clear RMQ after transforming.
> I hope to get your reply, Thanks a lot.
>
>
>
>
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
