You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by twalthr <gi...@git.apache.org> on 2015/10/07 14:59:55 UTC
[GitHub] flink pull request: [FLINK-2828] [table] Add interfaces for Table ...
GitHub user twalthr opened a pull request:
https://github.com/apache/flink/pull/1237
[FLINK-2828] [table] Add interfaces for Table API input formats
This PR implements TableSources as described in FLINK-2828. It is an uncoupling of #1127 for FLINK-2167. All feedback I got from @aljoscha is implemented.
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/twalthr/flink TableApiInputs
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/flink/pull/1237.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #1237
----
commit 67a0ac4af4dc3a063e699e6eed24a2c7177eb31e
Author: twalthr <tw...@apache.org>
Date: 2015-07-09T09:57:05Z
[FLINK-2828] [table] Add interfaces for Table API input formats
----
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-2828] [table] Add interfaces for Table ...
Posted by aljoscha <gi...@git.apache.org>.
Github user aljoscha commented on the pull request:
https://github.com/apache/flink/pull/1237#issuecomment-155500029
Yes, sorry for the delay.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-2828] [table] Add interfaces for Table ...
Posted by tillrohrmann <gi...@git.apache.org>.
Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/flink/pull/1237#discussion_r45600786
--- Diff: flink-staging/flink-table/src/main/scala/org/apache/flink/api/java/table/JavaBatchTranslator.scala ---
@@ -54,6 +55,22 @@ class JavaBatchTranslator extends PlanTranslator {
Table(Root(rowDataSet, resultFields))
}
+ override def createTable(tableSource: TableSource): Table = {
+ // a TableSource requires an ExecutionEnvironment
+ if (env.isEmpty) {
+ throw new ExpressionException("This operation requires an ExecutionEnvironment.")
+ }
+ tableSource match {
+ case adaptive: AdaptiveTableSource => Table(Root(adaptive, adaptive.getOutputFields()))
+
+ case static: StaticTableSource =>
+ createTable(static.createStaticDataSet(env.get),
--- End diff --
I think it's more scalaesque to use Scala's pattern matching to extract the `ExecutionEnvironment` from `env` instead of the `if` check with a following `get`.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-2828] [table] Add interfaces for Table ...
Posted by twalthr <gi...@git.apache.org>.
Github user twalthr commented on a diff in the pull request:
https://github.com/apache/flink/pull/1237#discussion_r43628710
--- Diff: flink-staging/flink-table/src/main/scala/org/apache/flink/api/table/expressions/analysis/FieldBacktracker.scala ---
@@ -0,0 +1,75 @@
+/*
+ * 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.api.table.expressions.analysis
+
+import org.apache.flink.api.table.expressions.{Naming, ResolvedFieldReference}
+import org.apache.flink.api.table.input.{AdaptiveTableSource, TableSource}
+import org.apache.flink.api.table.plan._
+
+object FieldBacktracker {
+
+ /**
+ * Tracks a field back to its Root and returns its original name and AdaptiveTableSource
+ * if possible.
+ * This only happens if the field is forwarded unmodified. Renaming operations are reverted.
+ *
+ * @param op start operator
+ * @param fieldName field name at start operator
+ * @return original field name with corresponding AdaptiveTableSource or null
+ */
+ def resolveFieldNameAndTableSource(op: PlanNode, fieldName: String):
+ (AdaptiveTableSource, String) = {
+ op match {
+ case s@Select(input, selection) =>
+ var resolvedField: (AdaptiveTableSource, String) = null
+ // only follow unmodified fields
+ selection.foreach {
--- End diff --
You are right the other resolved fields (even valid ones) were discarded and possibly overwritten by `None`, this was a bug. I changed the lines. Now the first resolved field found will be taken.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-2828] [table] Add interfaces for Table ...
Posted by aljoscha <gi...@git.apache.org>.
Github user aljoscha commented on the pull request:
https://github.com/apache/flink/pull/1237#issuecomment-155719774
Why is it `AdaptiveTableSource` and `StaticTableSource`? Aren't static sources a subset of adaptive sources that just don't adapt?
I don't think we need to have to interfaces and two ways of handling them in the translators. Or maybe there is another reason that I'm not seeing? :sweat_smile:
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-2828] [table] Add interfaces for Table ...
Posted by twalthr <gi...@git.apache.org>.
Github user twalthr commented on a diff in the pull request:
https://github.com/apache/flink/pull/1237#discussion_r43541584
--- Diff: flink-staging/flink-table/src/main/scala/org/apache/flink/api/table/expressions/analysis/FieldBacktracker.scala ---
@@ -0,0 +1,75 @@
+/*
+ * 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.api.table.expressions.analysis
+
+import org.apache.flink.api.table.expressions.{Naming, ResolvedFieldReference}
+import org.apache.flink.api.table.input.{AdaptiveTableSource, TableSource}
+import org.apache.flink.api.table.plan._
+
+object FieldBacktracker {
+
+ /**
+ * Tracks a field back to its Root and returns its original name and AdaptiveTableSource
+ * if possible.
+ * This only happens if the field is forwarded unmodified. Renaming operations are reverted.
+ *
+ * @param op start operator
+ * @param fieldName field name at start operator
+ * @return original field name with corresponding AdaptiveTableSource or null
+ */
+ def resolveFieldNameAndTableSource(op: PlanNode, fieldName: String):
+ (AdaptiveTableSource, String) = {
+ op match {
+ case s@Select(input, selection) =>
+ var resolvedField: (AdaptiveTableSource, String) = null
+ // only follow unmodified fields
+ selection.foreach {
--- End diff --
No, `selection` can have multiple elements. The lines above just make sure that at least one Expression links to the `TableSource`'s field.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-2828] [table] Add interfaces for Table ...
Posted by aljoscha <gi...@git.apache.org>.
Github user aljoscha commented on the pull request:
https://github.com/apache/flink/pull/1237#issuecomment-159215435
I think the performance of the pre-flight phase is mostly irrelevant, even if it is slow no-one will notice since it only runs once before the job starts. IMHO, simplicity is key here, and the two different sources require two code-paths to handle them even though the one source provides a subset of the functionality of the other source.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-2828] [table] Add interfaces for Table ...
Posted by twalthr <gi...@git.apache.org>.
Github user twalthr commented on a diff in the pull request:
https://github.com/apache/flink/pull/1237#discussion_r43742661
--- Diff: flink-staging/flink-table/src/main/scala/org/apache/flink/api/table/expressions/analysis/FieldBacktracker.scala ---
@@ -0,0 +1,79 @@
+/*
+ * 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.api.table.expressions.analysis
+
+import org.apache.flink.api.table.expressions.{Naming, ResolvedFieldReference}
+import org.apache.flink.api.table.input.{AdaptiveTableSource, TableSource}
+import org.apache.flink.api.table.plan._
+
+object FieldBacktracker {
+
+ /**
+ * Tracks a field back to its Root and returns its original name and AdaptiveTableSource
+ * if possible.
+ * This only happens if the field is forwarded unmodified. Renaming operations are reverted.
+ *
+ * @param op start operator
+ * @param fieldName field name at start operator
+ * @return original field name with corresponding AdaptiveTableSource
+ */
+ def resolveFieldNameAndTableSource(op: PlanNode, fieldName: String):
+ Option[(AdaptiveTableSource, String)] = {
+ op match {
+ case s@Select(input, selection) =>
+ var resolvedField: Option[(AdaptiveTableSource, String)] = None
+ // only follow unmodified fields
+ selection.foreach(expr => {
+ if (resolvedField.isEmpty) {
+ expr match {
+ case ResolvedFieldReference(name, _) if name == fieldName =>
+ resolvedField = resolveFieldNameAndTableSource(input, fieldName)
+ case n@Naming(ResolvedFieldReference(oldName, _), newName) if newName == fieldName =>
+ resolvedField = resolveFieldNameAndTableSource(input, oldName)
+ case _ => // do nothing
+ }
+ }
+ })
--- End diff --
Sorry, I wasn't aware of the non-strict/strict concepts in Scala so far. Your solution is definitely nicer. I will correct that ;)
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-2828] [table] Add interfaces for Table ...
Posted by tillrohrmann <gi...@git.apache.org>.
Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/flink/pull/1237#discussion_r42603297
--- Diff: flink-staging/flink-table/src/main/scala/org/apache/flink/api/table/expressions/analysis/FieldBacktracker.scala ---
@@ -0,0 +1,75 @@
+/*
+ * 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.api.table.expressions.analysis
+
+import org.apache.flink.api.table.expressions.{Naming, ResolvedFieldReference}
+import org.apache.flink.api.table.input.{AdaptiveTableSource, TableSource}
+import org.apache.flink.api.table.plan._
+
+object FieldBacktracker {
+
+ /**
+ * Tracks a field back to its Root and returns its original name and AdaptiveTableSource
+ * if possible.
+ * This only happens if the field is forwarded unmodified. Renaming operations are reverted.
+ *
+ * @param op start operator
+ * @param fieldName field name at start operator
+ * @return original field name with corresponding AdaptiveTableSource or null
+ */
+ def resolveFieldNameAndTableSource(op: PlanNode, fieldName: String):
+ (AdaptiveTableSource, String) = {
+ op match {
+ case s@Select(input, selection) =>
+ var resolvedField: (AdaptiveTableSource, String) = null
--- End diff --
Why not returning the resolved field directly from the `selection`?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-2828] [table] Add interfaces for Table ...
Posted by twalthr <gi...@git.apache.org>.
Github user twalthr commented on a diff in the pull request:
https://github.com/apache/flink/pull/1237#discussion_r43872771
--- Diff: flink-staging/flink-table/src/main/scala/org/apache/flink/api/table/expressions/analysis/FieldBacktracker.scala ---
@@ -0,0 +1,79 @@
+/*
+ * 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.api.table.expressions.analysis
+
+import org.apache.flink.api.table.expressions.{Naming, ResolvedFieldReference}
+import org.apache.flink.api.table.input.{AdaptiveTableSource, TableSource}
+import org.apache.flink.api.table.plan._
+
+object FieldBacktracker {
+
+ /**
+ * Tracks a field back to its Root and returns its original name and AdaptiveTableSource
+ * if possible.
+ * This only happens if the field is forwarded unmodified. Renaming operations are reverted.
+ *
+ * @param op start operator
+ * @param fieldName field name at start operator
+ * @return original field name with corresponding AdaptiveTableSource
+ */
+ def resolveFieldNameAndTableSource(op: PlanNode, fieldName: String):
+ Option[(AdaptiveTableSource, String)] = {
+ op match {
+ case s@Select(input, selection) =>
+ var resolvedField: Option[(AdaptiveTableSource, String)] = None
+ // only follow unmodified fields
+ selection.foreach(expr => {
+ if (resolvedField.isEmpty) {
+ expr match {
+ case ResolvedFieldReference(name, _) if name == fieldName =>
+ resolvedField = resolveFieldNameAndTableSource(input, fieldName)
+ case n@Naming(ResolvedFieldReference(oldName, _), newName) if newName == fieldName =>
+ resolvedField = resolveFieldNameAndTableSource(input, oldName)
+ case _ => // do nothing
+ }
+ }
+ })
--- End diff --
Build was successful (at least my parts). Do you think I can merge it?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-2828] [table] Add interfaces for Table ...
Posted by tillrohrmann <gi...@git.apache.org>.
Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/flink/pull/1237#discussion_r45602245
--- Diff: flink-staging/flink-table/src/main/scala/org/apache/flink/api/table/expressions/analysis/PredicatePruner.scala ---
@@ -0,0 +1,88 @@
+/*
+ * 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.api.table.expressions.analysis
+
+import org.apache.flink.api.table.expressions._
+
+object PredicatePruner {
+
+ /**
+ * Prunes an expression according to the given filter.
+ * One part of a conjunction remains if the other part is filtered out.
+ * A disjunction is removed if one part is filtered out.
+ *
+ * @param filter filter function
+ * @param expr expression to be copied and pruned
+ * @return new pruned expression
+ */
+ def pruneExpr(filter: Expression => Boolean, expr: Expression): Expression = {
+ expr match {
+ case And(left, right) =>
+ val prunedLeft = pruneExpr(filter, left)
+ val prunedRight = pruneExpr(filter, right)
+
+ if (prunedLeft.isInstanceOf[NopExpression]) {
+ prunedRight
+ }
+ else if (prunedRight.isInstanceOf[NopExpression]) {
+ prunedLeft
+ }
+ else {
+ And(prunedLeft, prunedRight)
+ }
--- End diff --
What if left and right are `NopExpressions`?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-2828] [table] Add interfaces for Table ...
Posted by aljoscha <gi...@git.apache.org>.
Github user aljoscha commented on a diff in the pull request:
https://github.com/apache/flink/pull/1237#discussion_r44515720
--- Diff: flink-scala/src/main/scala/org/apache/flink/api/scala/ExecutionEnvironment.scala ---
@@ -64,15 +62,19 @@ import scala.reflect.ClassTag
* created. If the program is submitted to a cluster a remote execution environment will
* be created.
*/
-class ExecutionEnvironment(javaEnv: JavaEnv) {
+class ExecutionEnvironment(javaEnv: JavaEnv) extends AbstractExecutionEnvironment {
/**
- * @return the Java Execution environment.
+ * Returns the enclosed Java ExecutionEnvironment for special use cases.
+ *
+ * @return reference to the ExecutionEnvironment of the Java API
*/
def getJavaEnv: JavaEnv = javaEnv
+
/**
* Gets the config object.
*/
+ @Override
--- End diff --
In Scala override is done as:
```
override def getConfig ...
```
this is also in StreamExecutionEnvironment.scala
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-2828] [table] Add interfaces for Table ...
Posted by twalthr <gi...@git.apache.org>.
Github user twalthr closed the pull request at:
https://github.com/apache/flink/pull/1237
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-2828] [table] Add interfaces for Table ...
Posted by tillrohrmann <gi...@git.apache.org>.
Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/flink/pull/1237#discussion_r42603082
--- Diff: flink-staging/flink-table/src/main/scala/org/apache/flink/api/table/expressions/analysis/FieldBacktracker.scala ---
@@ -0,0 +1,75 @@
+/*
+ * 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.api.table.expressions.analysis
+
+import org.apache.flink.api.table.expressions.{Naming, ResolvedFieldReference}
+import org.apache.flink.api.table.input.{AdaptiveTableSource, TableSource}
+import org.apache.flink.api.table.plan._
+
+object FieldBacktracker {
+
+ /**
+ * Tracks a field back to its Root and returns its original name and AdaptiveTableSource
+ * if possible.
+ * This only happens if the field is forwarded unmodified. Renaming operations are reverted.
+ *
+ * @param op start operator
+ * @param fieldName field name at start operator
+ * @return original field name with corresponding AdaptiveTableSource or null
+ */
+ def resolveFieldNameAndTableSource(op: PlanNode, fieldName: String):
+ (AdaptiveTableSource, String) = {
--- End diff --
Can we avoid `null` return values by using `Option[(AdaptiveTableSource, String)]`?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-2828] [table] Add interfaces for Table ...
Posted by twalthr <gi...@git.apache.org>.
Github user twalthr commented on the pull request:
https://github.com/apache/flink/pull/1237#issuecomment-158955674
@aljoscha I thought it makes sense to distinguish between `AdaptiveTableSource` and `StaticTableSource`. Calculating the predicates to be pushed down is quite expensive, it should only happen if the TableSource needs it. `AdaptiveTableSource` need to return `Rows` where fields can be null. `StaticTableSource`s can return POJOs or Tuples.
@tillrohrmann thanks again for your feedback, I will correct it.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-2828] [table] Add interfaces for Table ...
Posted by twalthr <gi...@git.apache.org>.
Github user twalthr commented on the pull request:
https://github.com/apache/flink/pull/1237#issuecomment-162622155
So basically what you want to say is: throw your entire code away.
I'm disappointed that you think about the whole concept now, after I have rebased and reworked the code multiple times. I published a prototype of it in mid-September (#1127), where I wanted to know if the concept is fine. I won't have the resources and motivation to start from scratch.
Yes, you are right. Your solution is the nicer one. However, the question is how much optimizer logic the Table API should have implemented. Actually I wanted to discuss the following under FLINK-2099, but we can also discuss it here. If we base some further developed library such as Calcite on top of the Table API, optimizer logic will be unnecessary in the near future. In a long term perspective it might make sense to convert Table API DSL to Calcite nodes, optimize them, and use the Table API plan nodes only as the physical plan nodes without further optimization. In this case the only thing missing is predicate/projection pushdown from the Table API tree into the input formats. Calcite could do push down for us.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-2828] [table] Add interfaces for Table ...
Posted by twalthr <gi...@git.apache.org>.
Github user twalthr commented on the pull request:
https://github.com/apache/flink/pull/1237#issuecomment-206214656
I will close this issue in favor of the new design `TableDef` design.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-2828] [table] Add interfaces for Table ...
Posted by aljoscha <gi...@git.apache.org>.
Github user aljoscha commented on the pull request:
https://github.com/apache/flink/pull/1237#issuecomment-153708732
I want to have a final look today. Is the Scala ok from your side now, @tillrohrmann?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-2828] [table] Add interfaces for Table ...
Posted by tillrohrmann <gi...@git.apache.org>.
Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/flink/pull/1237#discussion_r43590351
--- Diff: flink-staging/flink-table/src/main/scala/org/apache/flink/api/table/expressions/analysis/FieldBacktracker.scala ---
@@ -0,0 +1,75 @@
+/*
+ * 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.api.table.expressions.analysis
+
+import org.apache.flink.api.table.expressions.{Naming, ResolvedFieldReference}
+import org.apache.flink.api.table.input.{AdaptiveTableSource, TableSource}
+import org.apache.flink.api.table.plan._
+
+object FieldBacktracker {
+
+ /**
+ * Tracks a field back to its Root and returns its original name and AdaptiveTableSource
+ * if possible.
+ * This only happens if the field is forwarded unmodified. Renaming operations are reverted.
+ *
+ * @param op start operator
+ * @param fieldName field name at start operator
+ * @return original field name with corresponding AdaptiveTableSource or null
+ */
+ def resolveFieldNameAndTableSource(op: PlanNode, fieldName: String):
+ (AdaptiveTableSource, String) = {
+ op match {
+ case s@Select(input, selection) =>
+ var resolvedField: (AdaptiveTableSource, String) = null
+ // only follow unmodified fields
+ selection.foreach {
--- End diff --
But with this implementation you will take the resolved value of the last expression in `selection`. All other resolved fields will be discarded. Why don't you simply take the first element of `selection` via `head` and resolve it? Or do you perform any side effects in the resolve methods?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-2828] [table] Add interfaces for Table ...
Posted by tillrohrmann <gi...@git.apache.org>.
Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/flink/pull/1237#discussion_r45600975
--- Diff: flink-staging/flink-table/src/main/scala/org/apache/flink/api/java/table/JavaStreamingTranslator.scala ---
@@ -113,8 +131,14 @@ class JavaStreamingTranslator extends PlanTranslator {
private def translateInternal(op: PlanNode): DataStream[Row] = {
op match {
- case Root(dataSet: DataStream[Row], resultFields) =>
- dataSet
+ case Root(dataStream: DataStream[Row], resultFields) =>
+ dataStream
+
+ case Root(tableSource: AdaptiveTableSource, resultFields) =>
+ if (env.isEmpty) {
+ throw new ExpressionException("This operation requires a TableEnvironment.")
+ }
+ tableSource.createAdaptiveDataStream(env.get)
--- End diff --
`env` extraction
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-2828] [table] Add interfaces for Table ...
Posted by tillrohrmann <gi...@git.apache.org>.
Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/flink/pull/1237#discussion_r45600839
--- Diff: flink-staging/flink-table/src/main/scala/org/apache/flink/api/java/table/JavaBatchTranslator.scala ---
@@ -118,6 +135,12 @@ class JavaBatchTranslator extends PlanTranslator {
case Root(dataSet: JavaDataSet[Row], resultFields) =>
dataSet
+ case Root(tableSource: AdaptiveTableSource, resultFields) =>
+ if (env.isEmpty) {
+ throw new ExpressionException("This operation requires a TableEnvironment.")
+ }
+ tableSource.createAdaptiveDataSet(env.get)
--- End diff --
Same here.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-2828] [table] Add interfaces for Table ...
Posted by twalthr <gi...@git.apache.org>.
Github user twalthr commented on the pull request:
https://github.com/apache/flink/pull/1237#issuecomment-162521847
@aljoscha we should finally reach a consesus for this PR
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-2828] [table] Add interfaces for Table ...
Posted by twalthr <gi...@git.apache.org>.
Github user twalthr commented on the pull request:
https://github.com/apache/flink/pull/1237#issuecomment-159229093
Most input formats won't use the adaptive sources. Sure performance is not that important in the pre-flight phase, but if we always think so, the pre-flight phase will some day take minutes. Simplicity is key but IMHO it is simpler from an extension point of view to implement an interface that fits to the need. One type is evaluated immediately the other is lazy evaluated during translation. I think this is differentiation/argument enough for 2 interfaces.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-2828] [table] Add interfaces for Table ...
Posted by aljoscha <gi...@git.apache.org>.
Github user aljoscha commented on the pull request:
https://github.com/apache/flink/pull/1237#issuecomment-205881346
@twalthr is this still valid with the changes in the Table API or can it be closed?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-2828] [table] Add interfaces for Table ...
Posted by tillrohrmann <gi...@git.apache.org>.
Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/flink/pull/1237#discussion_r45601100
--- Diff: flink-staging/flink-table/src/main/scala/org/apache/flink/api/java/table/TableEnvironment.scala ---
@@ -45,7 +60,12 @@ class TableEnvironment {
* are named a and b.
*/
def fromDataSet[T](set: DataSet[T], fields: String): Table = {
- new JavaBatchTranslator().createTable(set, fields)
+ translatorFromEnv match {
+ case batchTranslator: JavaBatchTranslator =>
+ batchTranslator.createTable(set, fields)
+ case _ =>
+ throw new IllegalArgumentException("ExecutionEnvironment does not support Java DataSets.")
--- End diff --
Shouldn't it be `DataStreams` instead of `DataSets`?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-2828] [table] Add interfaces for Table ...
Posted by tillrohrmann <gi...@git.apache.org>.
Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/flink/pull/1237#discussion_r42602668
--- Diff: flink-staging/flink-table/src/main/scala/org/apache/flink/api/java/table/JavaStreamingTranslator.scala ---
@@ -38,7 +40,7 @@ import org.apache.flink.streaming.api.operators.StreamMap
* operations must be extended to allow windowing operations.
*/
-class JavaStreamingTranslator extends PlanTranslator {
+class JavaStreamingTranslator(env: StreamExecutionEnvironment = null) extends PlanTranslator {
--- End diff --
same here
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-2828] [table] Add interfaces for Table ...
Posted by tillrohrmann <gi...@git.apache.org>.
Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/flink/pull/1237#discussion_r42602646
--- Diff: flink-staging/flink-table/src/main/scala/org/apache/flink/api/java/table/JavaBatchTranslator.scala ---
@@ -39,7 +40,7 @@ import org.apache.flink.api.table.{ExpressionException, Row, Table}
* [[PlanTranslator]] for creating [[Table]]s from Java [[org.apache.flink.api.java.DataSet]]s and
* translating them back to Java [[org.apache.flink.api.java.DataSet]]s.
*/
-class JavaBatchTranslator extends PlanTranslator {
+class JavaBatchTranslator(env: ExecutionEnvironment = null) extends PlanTranslator {
--- End diff --
We could use an `Option` here to avoid `null`, since `null` is evil.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-2828] [table] Add interfaces for Table ...
Posted by twalthr <gi...@git.apache.org>.
Github user twalthr commented on the pull request:
https://github.com/apache/flink/pull/1237#issuecomment-155498508
@aljoscha Do you think you'll find time this week for the PR? All Table API input formats are waiting for it...
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-2828] [table] Add interfaces for Table ...
Posted by aljoscha <gi...@git.apache.org>.
Github user aljoscha commented on the pull request:
https://github.com/apache/flink/pull/1237#issuecomment-162654340
You are right of course and I'm sorry I didn't put in the required time earlier.
Your are also right about the Calcite thing. If we decide to use calcite and have the Table API basically as another fronted to generate SQL plans then the Table API optimization part can be very thin. In that case we would also not need the Predicate push down as implemented as part of this PR, though, and can keep it lean and simple.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-2828] [table] Add interfaces for Table ...
Posted by tillrohrmann <gi...@git.apache.org>.
Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/flink/pull/1237#discussion_r42603373
--- Diff: flink-staging/flink-table/src/main/scala/org/apache/flink/api/table/expressions/analysis/PredicatePruner.scala ---
@@ -0,0 +1,88 @@
+/*
+ * 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.api.table.expressions.analysis
+
+import org.apache.flink.api.table.expressions._
+
+object PredicatePruner {
+
+ /**
+ * Prunes an expression according to the given filter.
+ * One part of a conjunction remains if the other part is filtered out.
+ * A disjunction is removed if one part is filtered out.
+ *
+ * @param filter filter function
+ * @param expr expression to be copied and pruned
+ * @return new pruned expression
+ */
+ def pruneExpr(filter: (Expression) => Boolean, expr: Expression): Expression = {
--- End diff --
Parentheses needed for `(Expression)`?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-2828] [table] Add interfaces for Table ...
Posted by twalthr <gi...@git.apache.org>.
Github user twalthr commented on a diff in the pull request:
https://github.com/apache/flink/pull/1237#discussion_r43542701
--- Diff: flink-staging/flink-table/src/main/scala/org/apache/flink/api/table/expressions/analysis/FieldBacktracker.scala ---
@@ -0,0 +1,75 @@
+/*
+ * 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.api.table.expressions.analysis
+
+import org.apache.flink.api.table.expressions.{Naming, ResolvedFieldReference}
+import org.apache.flink.api.table.input.{AdaptiveTableSource, TableSource}
+import org.apache.flink.api.table.plan._
+
+object FieldBacktracker {
+
+ /**
+ * Tracks a field back to its Root and returns its original name and AdaptiveTableSource
+ * if possible.
+ * This only happens if the field is forwarded unmodified. Renaming operations are reverted.
+ *
+ * @param op start operator
+ * @param fieldName field name at start operator
+ * @return original field name with corresponding AdaptiveTableSource or null
+ */
+ def resolveFieldNameAndTableSource(op: PlanNode, fieldName: String):
+ (AdaptiveTableSource, String) = {
+ op match {
+ case s@Select(input, selection) =>
+ var resolvedField: (AdaptiveTableSource, String) = null
--- End diff --
See answer above.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-2828] [table] Add interfaces for Table ...
Posted by aljoscha <gi...@git.apache.org>.
Github user aljoscha commented on the pull request:
https://github.com/apache/flink/pull/1237#issuecomment-162553706
Yes, we should, this has been lying around for a while now.
I have been thinking about this for a while now and I'm afraid I cannot recommend to merge this in the current form. This PR is tackling two issues: 1) adding new interfaces for Table API input formats, 2) pushing filter predicates into sources. 1) will end up being a question of style, i.e. whether we want to have two input interfaces, or just one that can do both.
On 2), I think we need a different approach. The approach taken in this PR is to enhance the Rules that analyze/transform expression with knowledge about the input operation. This allows to backtrack fields to the sources and notify the source of used fields/predicates. I think, however, that this is not the correct level to implement it. The plan of operations is a Tree itself and can also be analyzed/transformed using the Tree framework of the Table API. I think it would be beneficial to provide analysis steps that can actually push filters towards the sources. (And then fuse adaptive source and filter if possible) This will be helpful even if no adaptive sources are present and it will transform one valid tree of the user query into another valid tree of the user query which makes it easy to see the optimizations that are taking place and to debug it.
To show what I mean I implemented a small prototype that prunes unused fields of selection expressions and pushes filters towards sources: https://github.com/aljoscha/flink/tree/table-filter-push. There is an example in there: `FilterPushExample.scala`, the code for tree analysis is in `PlanAnalyzer.scala`. The example query is this:
```
val input = env.fromElements((1, 1, 1), (2, 2, 2), (4, 4, 4), (5, 5, 5))
input
.as('a, 'b, 'c)
.select('a as 'd, 'b, 'c + 15 as 'e)
.filter('d > 3)
.select('d)
.toDataSet[Row].print
```
if the example is run it will print the operation tree after tree analysis phases, this is the printout:
```
Before PushFilter: Select(Filter(Select(Root(MapOperator@49c43f4e,ArraySeq((a,Integer), (b,Integer), (c,Integer))), 'a as 'd,'b,('c + 15) as 'e), 'd > 3), 'd)
After PushFilter: Select(Select(Filter(Root(MapOperator@49c43f4e,ArraySeq((a,Integer), (b,Integer), (c,Integer))), 'a > 3), 'a as 'd,'b,('c + 15) as 'e), 'd)
Before prune: Select(Select(Filter(Root(MapOperator@49c43f4e,ArraySeq((a,Integer), (b,Integer), (c,Integer))), 'a > 3), 'a as 'd,'b,('c + 15) as 'e), 'd)
After prune: Select(Select(Filter(Root(MapOperator@49c43f4e,ArraySeq((a,Integer), (b,Integer), (c,Integer))), 'a > 3), 'a as 'd), 'd)
ROOT used fields: a
```
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-2828] [table] Add interfaces for Table ...
Posted by tillrohrmann <gi...@git.apache.org>.
Github user tillrohrmann commented on the pull request:
https://github.com/apache/flink/pull/1237#issuecomment-153709718
Yeah looks good now :-)
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-2828] [table] Add interfaces for Table ...
Posted by tillrohrmann <gi...@git.apache.org>.
Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/flink/pull/1237#discussion_r45601201
--- Diff: flink-staging/flink-table/src/main/scala/org/apache/flink/api/java/table/TableEnvironment.scala ---
@@ -45,7 +60,12 @@ class TableEnvironment {
* are named a and b.
*/
def fromDataSet[T](set: DataSet[T], fields: String): Table = {
- new JavaBatchTranslator().createTable(set, fields)
+ translatorFromEnv match {
+ case batchTranslator: JavaBatchTranslator =>
+ batchTranslator.createTable(set, fields)
+ case _ =>
+ throw new IllegalArgumentException("ExecutionEnvironment does not support Java DataSets.")
--- End diff --
Sorry, forget my comment.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-2828] [table] Add interfaces for Table ...
Posted by twalthr <gi...@git.apache.org>.
Github user twalthr commented on a diff in the pull request:
https://github.com/apache/flink/pull/1237#discussion_r43541130
--- Diff: flink-staging/flink-table/src/main/scala/org/apache/flink/api/java/table/JavaBatchTranslator.scala ---
@@ -39,7 +40,7 @@ import org.apache.flink.api.table.{ExpressionException, Row, Table}
* [[PlanTranslator]] for creating [[Table]]s from Java [[org.apache.flink.api.java.DataSet]]s and
* translating them back to Java [[org.apache.flink.api.java.DataSet]]s.
*/
-class JavaBatchTranslator extends PlanTranslator {
+class JavaBatchTranslator(env: ExecutionEnvironment = null) extends PlanTranslator {
--- End diff --
You are right. Sorry, Scala beginner's mistake. I will correct that.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-2828] [table] Add interfaces for Table ...
Posted by tillrohrmann <gi...@git.apache.org>.
Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/flink/pull/1237#discussion_r42603257
--- Diff: flink-staging/flink-table/src/main/scala/org/apache/flink/api/table/expressions/analysis/FieldBacktracker.scala ---
@@ -0,0 +1,75 @@
+/*
+ * 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.api.table.expressions.analysis
+
+import org.apache.flink.api.table.expressions.{Naming, ResolvedFieldReference}
+import org.apache.flink.api.table.input.{AdaptiveTableSource, TableSource}
+import org.apache.flink.api.table.plan._
+
+object FieldBacktracker {
+
+ /**
+ * Tracks a field back to its Root and returns its original name and AdaptiveTableSource
+ * if possible.
+ * This only happens if the field is forwarded unmodified. Renaming operations are reverted.
+ *
+ * @param op start operator
+ * @param fieldName field name at start operator
+ * @return original field name with corresponding AdaptiveTableSource or null
+ */
+ def resolveFieldNameAndTableSource(op: PlanNode, fieldName: String):
+ (AdaptiveTableSource, String) = {
+ op match {
+ case s@Select(input, selection) =>
+ var resolvedField: (AdaptiveTableSource, String) = null
+ // only follow unmodified fields
+ selection.foreach {
--- End diff --
`selection` is of type `Seq[Expression]`, right? But `resolveField` stores only the last resolved field. Is this correct? Shouldn't we make sure that the selection contains only a single element?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-2828] [table] Add interfaces for Table ...
Posted by tillrohrmann <gi...@git.apache.org>.
Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/flink/pull/1237#discussion_r45602715
--- Diff: flink-staging/flink-table/src/main/scala/org/apache/flink/api/table/expressions/analysis/PredicatePushdown.scala ---
@@ -0,0 +1,120 @@
+/*
+ * 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.api.table.expressions.analysis
+
+import org.apache.flink.api.table.expressions._
+import org.apache.flink.api.table.expressions.analysis.FieldBacktracker
+ .resolveFieldNameAndTableSource
+import org.apache.flink.api.table.expressions.analysis.PredicatePruner.pruneExpr
+import org.apache.flink.api.table.input.AdaptiveTableSource
+import org.apache.flink.api.table.plan._
+import org.apache.flink.api.table.trees.Rule
+
+/**
+ * Pushes constant predicates (e.g. a===12 && b.isNotNull) to each corresponding
+ * AdaptiveTableSource that support predicates.
+ */
+class PredicatePushdown(val inputOperation: PlanNode) extends Rule[Expression] {
+
+ def apply(expr: Expression) = {
+ // get all table sources where predicates can be push into
+ val tableSources = getPushableTableSources(inputOperation)
+
+ // prune expression tree such that it only contains constant predicates
+ // such as a=1,a="Hello World", isNull(a) but not a=b
+ val constantExpr = pruneExpr(isResolvedAndConstant, expr)
+
+ // push predicates to each table source respectively
+ for (ts <- tableSources) {
+ // prune expression tree such that it only contains field references of ts
+ val tsExpr = pruneExpr((e) => isSameTableSource(e, ts), constantExpr)
+
+ // resolve field names to field names of the table source
+ val result = tsExpr.transformPost {
+ case rfr@ResolvedFieldReference(fieldName, typeInfo) =>
+ ResolvedFieldReference(
+ resolveFieldNameAndTableSource(inputOperation, fieldName).get._2,
+ typeInfo
+ )
+ }
+ // push down predicates
+ if (result != NopExpression()) {
+ ts.notifyPredicates(result)
+ }
+ }
+ expr
+ }
+
+ // ----------------------------------------------------------------------------------------------
+
+ /**
+ * @return all AdaptiveTableSources the given PlanNode contains
+ */
+ def getPushableTableSources(tree: PlanNode): Seq[AdaptiveTableSource] = tree match {
+ case Root(ts: AdaptiveTableSource, _) => Seq(ts)
+ case pn: PlanNode =>
+ pn.children flatMap { child => getPushableTableSources(child ) }
+ case _ => Seq() // add nothing
+ }
+
+ /**
+ *
+ * @return true if the given expression is a predicate that consists of a
+ * ResolvedFieldReference and a constant/literal
+ * e.g. a=1, 2<c
+ */
+ def isResolvedAndConstant(expr: Expression) : Boolean = {
+ expr match {
+ case bc: BinaryComparison if bc.left.isInstanceOf[Literal]
+ && bc.right.isInstanceOf[ResolvedFieldReference] =>
+ true
+ case bc: BinaryComparison if bc.right.isInstanceOf[Literal]
+ && bc.left.isInstanceOf[ResolvedFieldReference] =>
+ true
+ case ue@(IsNotNull(_) | IsNull(_)) =>
+ val child = ue.asInstanceOf[UnaryExpression].child
+ child.isInstanceOf[ResolvedFieldReference]
+ case And(_,_) | Or(_,_) | Not(_) =>
+ true
+ case _ => false
+ }
+ }
+
+ /**
+ * @return true if the given expression only consists of ResolvedFieldReference of
+ * the same given AdaptiveTableSource
+ */
+ def isSameTableSource(expr: Expression, ts: AdaptiveTableSource) : Boolean = {
+ expr match {
+ case bc: BinaryComparison if bc.right.isInstanceOf[ResolvedFieldReference] =>
+ val fieldRef = bc.right.asInstanceOf[ResolvedFieldReference]
+ val resolvedField = resolveFieldNameAndTableSource(inputOperation, fieldRef.name)
+ resolvedField.isDefined && resolvedField.get._1 == ts
--- End diff --
pattern matching is more scalaesque
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-2828] [table] Add interfaces for Table ...
Posted by tillrohrmann <gi...@git.apache.org>.
Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/flink/pull/1237#discussion_r45600858
--- Diff: flink-staging/flink-table/src/main/scala/org/apache/flink/api/java/table/JavaStreamingTranslator.scala ---
@@ -53,6 +55,22 @@ class JavaStreamingTranslator extends PlanTranslator {
new Table(Root(rowDataStream, resultFields))
}
+ override def createTable(tableSource: TableSource): Table = {
+ // a TableSource requires an StreamExecutionEnvironment
+ if (env.isEmpty) {
+ throw new ExpressionException("This operation requires a StreamExecutionEnvironment.")
+ }
+ tableSource match {
+ case adaptive: AdaptiveTableSource => Table(Root(adaptive, adaptive.getOutputFields()))
+
+ case static: StaticTableSource =>
+ createTable(static.createStaticDataStream(env.get),
--- End diff --
here as well
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-2828] [table] Add interfaces for Table ...
Posted by tillrohrmann <gi...@git.apache.org>.
Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/flink/pull/1237#discussion_r45602446
--- Diff: flink-staging/flink-table/src/main/scala/org/apache/flink/api/table/expressions/analysis/PredicatePushdown.scala ---
@@ -0,0 +1,120 @@
+/*
+ * 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.api.table.expressions.analysis
+
+import org.apache.flink.api.table.expressions._
+import org.apache.flink.api.table.expressions.analysis.FieldBacktracker
+ .resolveFieldNameAndTableSource
+import org.apache.flink.api.table.expressions.analysis.PredicatePruner.pruneExpr
+import org.apache.flink.api.table.input.AdaptiveTableSource
+import org.apache.flink.api.table.plan._
+import org.apache.flink.api.table.trees.Rule
+
+/**
+ * Pushes constant predicates (e.g. a===12 && b.isNotNull) to each corresponding
+ * AdaptiveTableSource that support predicates.
+ */
+class PredicatePushdown(val inputOperation: PlanNode) extends Rule[Expression] {
+
+ def apply(expr: Expression) = {
+ // get all table sources where predicates can be push into
+ val tableSources = getPushableTableSources(inputOperation)
+
+ // prune expression tree such that it only contains constant predicates
+ // such as a=1,a="Hello World", isNull(a) but not a=b
+ val constantExpr = pruneExpr(isResolvedAndConstant, expr)
+
+ // push predicates to each table source respectively
+ for (ts <- tableSources) {
+ // prune expression tree such that it only contains field references of ts
+ val tsExpr = pruneExpr((e) => isSameTableSource(e, ts), constantExpr)
+
+ // resolve field names to field names of the table source
+ val result = tsExpr.transformPost {
+ case rfr@ResolvedFieldReference(fieldName, typeInfo) =>
+ ResolvedFieldReference(
+ resolveFieldNameAndTableSource(inputOperation, fieldName).get._2,
--- End diff --
Are we sure that this `Option` is always `Some`?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-2828] [table] Add interfaces for Table ...
Posted by tillrohrmann <gi...@git.apache.org>.
Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/flink/pull/1237#discussion_r43631877
--- Diff: flink-staging/flink-table/src/main/scala/org/apache/flink/api/table/expressions/analysis/FieldBacktracker.scala ---
@@ -0,0 +1,79 @@
+/*
+ * 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.api.table.expressions.analysis
+
+import org.apache.flink.api.table.expressions.{Naming, ResolvedFieldReference}
+import org.apache.flink.api.table.input.{AdaptiveTableSource, TableSource}
+import org.apache.flink.api.table.plan._
+
+object FieldBacktracker {
+
+ /**
+ * Tracks a field back to its Root and returns its original name and AdaptiveTableSource
+ * if possible.
+ * This only happens if the field is forwarded unmodified. Renaming operations are reverted.
+ *
+ * @param op start operator
+ * @param fieldName field name at start operator
+ * @return original field name with corresponding AdaptiveTableSource
+ */
+ def resolveFieldNameAndTableSource(op: PlanNode, fieldName: String):
+ Option[(AdaptiveTableSource, String)] = {
+ op match {
+ case s@Select(input, selection) =>
+ var resolvedField: Option[(AdaptiveTableSource, String)] = None
+ // only follow unmodified fields
+ selection.foreach(expr => {
+ if (resolvedField.isEmpty) {
+ expr match {
+ case ResolvedFieldReference(name, _) if name == fieldName =>
+ resolvedField = resolveFieldNameAndTableSource(input, fieldName)
+ case n@Naming(ResolvedFieldReference(oldName, _), newName) if newName == fieldName =>
+ resolvedField = resolveFieldNameAndTableSource(input, oldName)
+ case _ => // do nothing
+ }
+ }
+ })
--- End diff --
But here again you will traverse the whole `selection` sequence. Is there any special reason for it? Why not simply calculate the elements which are needed to find the first resolved field? With Scala you have all the possibilities at hand and furthermore you get rid of the `var` variable:
```
val resolvedField = selection.view.flatMap{
case ResolvedFieldReference(name, _) if name == fieldName =>
resolvedField = resolveFieldNameAndTableSource(input, fieldName)
case n@Naming(ResolvedFieldReference(oldName, _), newName) if newName == fieldName =>
resolvedField = resolveFieldNameAndTableSource(input, oldName)
case _ => // do nothing
}.headOption
```
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---