Repository: flink Updated Branches: refs/heads/master 770f2f83a -> 227cdc829
[FLINK-4581] [table] Fix Table API throwing "No suitable driver found for jdbc:calcite" This closes #2506 This closes #1491 // closing stale PR This closes #997 // closing stale PR Project: http://git-wip-us.apache.org/repos/asf/flink/repo Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/f0e451ae Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/f0e451ae Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/f0e451ae Branch: refs/heads/master Commit: f0e451ae3cbfe40926d08ffaf04cdff58136b306 Parents: 770f2f8 Author: twalthr <[email protected]> Authored: Fri Sep 16 11:41:15 2016 +0200 Committer: Fabian Hueske <[email protected]> Committed: Sat Oct 22 10:09:32 2016 +0200 ---------------------------------------------------------------------- .../flink/api/table/FlinkRelBuilder.scala | 38 ++++++-------------- 1 file changed, 11 insertions(+), 27 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/flink/blob/f0e451ae/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/FlinkRelBuilder.scala ---------------------------------------------------------------------- diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/FlinkRelBuilder.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/FlinkRelBuilder.scala index 34ed4ce..1215806 100644 --- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/FlinkRelBuilder.scala +++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/FlinkRelBuilder.scala @@ -18,24 +18,25 @@ package org.apache.flink.api.table +import java.util.Collections + import org.apache.calcite.jdbc.CalciteSchema -import org.apache.calcite.plan.{Context, RelOptCluster, RelOptPlanner, RelOptSchema} +import org.apache.calcite.plan._ +import org.apache.calcite.plan.volcano.VolcanoPlanner import org.apache.calcite.prepare.CalciteCatalogReader -import org.apache.calcite.rex.{RexExecutorImpl, RexBuilder} -import org.apache.calcite.schema.{Schemas, SchemaPlus} -import org.apache.calcite.tools.Frameworks.PlannerAction -import org.apache.calcite.tools.{FrameworkConfig, Frameworks, RelBuilder} +import org.apache.calcite.rex.RexBuilder +import org.apache.calcite.tools.{FrameworkConfig, RelBuilder} /** * Flink specific [[RelBuilder]] that changes the default type factory to a [[FlinkTypeFactory]]. */ class FlinkRelBuilder( context: Context, - cluster: RelOptCluster, + relOptCluster: RelOptCluster, relOptSchema: RelOptSchema) extends RelBuilder( context, - cluster, + relOptCluster, relOptSchema) { def getPlanner: RelOptPlanner = cluster.getPlanner @@ -49,37 +50,20 @@ class FlinkRelBuilder( object FlinkRelBuilder { def create(config: FrameworkConfig): FlinkRelBuilder = { - // prepare planner and collect context instances - val clusters: Array[RelOptCluster] = Array(null) - val relOptSchemas: Array[RelOptSchema] = Array(null) - val rootSchemas: Array[SchemaPlus] = Array(null) - Frameworks.withPlanner(new PlannerAction[Void] { - override def apply( - cluster: RelOptCluster, - relOptSchema: RelOptSchema, - rootSchema: SchemaPlus) - : Void = { - clusters(0) = cluster - relOptSchemas(0) = relOptSchema - rootSchemas(0) = rootSchema - null - } - }) - val planner = clusters(0).getPlanner - planner.setExecutor(config.getExecutor) - val defaultRelOptSchema = relOptSchemas(0).asInstanceOf[CalciteCatalogReader] // create Flink type factory val typeSystem = config.getTypeSystem val typeFactory = new FlinkTypeFactory(typeSystem) // create context instances with Flink type factory + val planner = new VolcanoPlanner(Contexts.empty()) + planner.addRelTraitDef(ConventionTraitDef.INSTANCE) val cluster = RelOptCluster.create(planner, new RexBuilder(typeFactory)) val calciteSchema = CalciteSchema.from(config.getDefaultSchema) val relOptSchema = new CalciteCatalogReader( calciteSchema, config.getParserConfig.caseSensitive(), - defaultRelOptSchema.getSchemaName, + Collections.emptyList(), typeFactory) new FlinkRelBuilder(config.getContext, cluster, relOptSchema)
