[jira] [Commented] (FLINK-2167) Add fromHCat() to TableEnvironment
[
https://issues.apache.org/jira/browse/FLINK-2167?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=14727333#comment-14727333
]
ASF GitHub Bot commented on FLINK-2167:
---
Github user twalthr commented on the pull request:
https://github.com/apache/flink/pull/1064#issuecomment-137080649
@jamescao: It seems that you also wrote tests for the HCatInputFormat,
right? is it possible to split the PR into a OutputFormat part and open a
separate PR for the HCatInputFormat tests. I'm still working on FLINK-2167 and
require a HCatalog testing infrastructure. Otherwise I have to write it my own.
Anyway, I wonder why all HCat I/O format classes have no tests so far...
> Add fromHCat() to TableEnvironment
> --
>
> Key: FLINK-2167
> URL: https://issues.apache.org/jira/browse/FLINK-2167
> Project: Flink
> Issue Type: New Feature
> Components: Table API
>Affects Versions: 0.9
>Reporter: Fabian Hueske
>Assignee: Timo Walther
>Priority: Minor
> Labels: starter
>
> Add a {{fromHCat()}} method to the {{TableEnvironment}} to read a {{Table}}
> from an HCatalog table.
> The implementation could reuse Flink's HCatInputFormat.
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
[jira] [Commented] (FLINK-2167) Add fromHCat() to TableEnvironment
[
https://issues.apache.org/jira/browse/FLINK-2167?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=14743517#comment-14743517
]
ASF GitHub Bot commented on FLINK-2167:
---
GitHub user twalthr opened a pull request:
https://github.com/apache/flink/pull/1127
[FLINK-2167] [table] Add fromHCat() to TableEnvironment
This PR introduces input format interfaces (so-called `TableSource`s) for
the Table API. There are two types of TableSources:
- `AdaptiveTableSource`s can adapt their output to the requirements of the
plan. Although the output schema stays the same, the TableSource can react on
field resolution and/or predicates internally and can return adapted
DataSet/DataStream versions in the "translate" step.
- `StaticTableSource`s are an easy way to provide the Table API with
additional input formats without much implementation effort (e.g. for
fromCsvFile())
TableSource have been deeply integrated into the Table API.
The TableEnvironment now requires the newly introduced
`AbstractExecutionEnvironment` (common super class of all ExecutionEnvironments
for DataSets and DataStreams).
An example of an AdaptiveTableSources can be found in `HCatTableSource`.
HCatTableSource supports predicate pushdown as well as selection pushdown to
HCatalog. Only those predicates are pushed to HCatalog that are partioned
columns. Unresolved fields will not be read from HCatalog and remain `null`
within the Table APIs rows.
A an easy example looks like:
```
TableEnironment t = new TableEnvironment(env);
t.fromHCat("database", "table")
.select("col1, col2")
.filter("partCol==='5'");
```
Here's what a TableSource can see from more complicated queries:
```
getTableJava(tableSource1)
.filter("a===5 || a===6")
.select("a as a4, b as b4, c as c4")
.filter("b4===7")
.join(getTableJava(tableSource2))
.where("a===a4 && c==='Test' && c4==='Test2'")
// Result predicates for tableSource1:
// List("a===5 || a===6", "b===7", "c==='Test2'")
// Result predicates for tableSource2:
// List("c==='Test'")
// Result resolved fields for tableSource1 (true = filtering,
false=selection):
// Set(("a", true), ("a", false), ("b", true), ("b", false), ("c", false),
("c", true))
// Result resolved fields for tableSource2 (true = filtering,
false=selection):
// Set(("a", true), ("c", true))
```
HCatTableSource has no tests yet, but I will implement it them soon. First
I would be happy about some general feedback.
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/twalthr/flink TableApiHcat
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/flink/pull/1127.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 #1127
commit f245604caccd8f97c1d6eabf16968dab3aa47572
Author: twalthr
Date: 2015-07-09T09:57:05Z
[FLINK-2167] [table] Add fromHCat() to TableEnvironment
> Add fromHCat() to TableEnvironment
> --
>
> Key: FLINK-2167
> URL: https://issues.apache.org/jira/browse/FLINK-2167
> Project: Flink
> Issue Type: New Feature
> Components: Table API
>Affects Versions: 0.9
>Reporter: Fabian Hueske
>Assignee: Timo Walther
>Priority: Minor
> Labels: starter
>
> Add a {{fromHCat()}} method to the {{TableEnvironment}} to read a {{Table}}
> from an HCatalog table.
> The implementation could reuse Flink's HCatInputFormat.
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
[jira] [Commented] (FLINK-2167) Add fromHCat() to TableEnvironment
[
https://issues.apache.org/jira/browse/FLINK-2167?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=14745078#comment-14745078
]
ASF GitHub Bot commented on FLINK-2167:
---
Github user chiwanpark commented on the pull request:
https://github.com/apache/flink/pull/1127#issuecomment-140327645
Hi @twalthr, thanks for your contribution. But this PR contains many
changes unrelated to HCatalog format. Maybe we should split this PR into
HCatalog and other changes.
> Add fromHCat() to TableEnvironment
> --
>
> Key: FLINK-2167
> URL: https://issues.apache.org/jira/browse/FLINK-2167
> Project: Flink
> Issue Type: New Feature
> Components: Table API
>Affects Versions: 0.9
>Reporter: Fabian Hueske
>Assignee: Timo Walther
>Priority: Minor
> Labels: starter
>
> Add a {{fromHCat()}} method to the {{TableEnvironment}} to read a {{Table}}
> from an HCatalog table.
> The implementation could reuse Flink's HCatInputFormat.
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
[jira] [Commented] (FLINK-2167) Add fromHCat() to TableEnvironment
[
https://issues.apache.org/jira/browse/FLINK-2167?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=14745260#comment-14745260
]
ASF GitHub Bot commented on FLINK-2167:
---
Github user twalthr commented on the pull request:
https://github.com/apache/flink/pull/1127#issuecomment-140355813
Hi @chiwanpark, yes I think splitting it up makes sense. I just opened this
PR to get some feedback and to show why my changes are necessary to integrate
new input formats like HCatalog. You can ignore the `HCatTableSource` class as
it is untested yet anyway.
> Add fromHCat() to TableEnvironment
> --
>
> Key: FLINK-2167
> URL: https://issues.apache.org/jira/browse/FLINK-2167
> Project: Flink
> Issue Type: New Feature
> Components: Table API
>Affects Versions: 0.9
>Reporter: Fabian Hueske
>Assignee: Timo Walther
>Priority: Minor
> Labels: starter
>
> Add a {{fromHCat()}} method to the {{TableEnvironment}} to read a {{Table}}
> from an HCatalog table.
> The implementation could reuse Flink's HCatInputFormat.
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
[jira] [Commented] (FLINK-2167) Add fromHCat() to TableEnvironment
[
https://issues.apache.org/jira/browse/FLINK-2167?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=14933090#comment-14933090
]
ASF GitHub Bot commented on FLINK-2167:
---
Github user aljoscha commented on a diff in the pull request:
https://github.com/apache/flink/pull/1127#discussion_r40534229
--- Diff: flink-staging/flink-table/pom.xml ---
@@ -59,6 +59,12 @@ under the License.
+ org.apache.flink
+ flink-hcatalog
+ ${project.version}
+
+
+
--- End diff --
Is it necessary to have the hcatalog dependency here or could it be put
into another package?
> Add fromHCat() to TableEnvironment
> --
>
> Key: FLINK-2167
> URL: https://issues.apache.org/jira/browse/FLINK-2167
> Project: Flink
> Issue Type: New Feature
> Components: Table API
>Affects Versions: 0.9
>Reporter: Fabian Hueske
>Assignee: Timo Walther
>Priority: Minor
> Labels: starter
>
> Add a {{fromHCat()}} method to the {{TableEnvironment}} to read a {{Table}}
> from an HCatalog table.
> The implementation could reuse Flink's HCatInputFormat.
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
[jira] [Commented] (FLINK-2167) Add fromHCat() to TableEnvironment
[
https://issues.apache.org/jira/browse/FLINK-2167?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=14933088#comment-14933088
]
ASF GitHub Bot commented on FLINK-2167:
---
Github user aljoscha commented on a diff in the pull request:
https://github.com/apache/flink/pull/1127#discussion_r40534193
--- Diff: flink-staging/flink-table/pom.xml ---
@@ -37,7 +37,7 @@ under the License.
com.google.guava
guava
- ${guava.version}
+ 14.0.1
--- End diff --
Why do you manually set the version here? Is it necessary because of
something with HCat? @rmetzger what could we do in such a case?
> Add fromHCat() to TableEnvironment
> --
>
> Key: FLINK-2167
> URL: https://issues.apache.org/jira/browse/FLINK-2167
> Project: Flink
> Issue Type: New Feature
> Components: Table API
>Affects Versions: 0.9
>Reporter: Fabian Hueske
>Assignee: Timo Walther
>Priority: Minor
> Labels: starter
>
> Add a {{fromHCat()}} method to the {{TableEnvironment}} to read a {{Table}}
> from an HCatalog table.
> The implementation could reuse Flink's HCatInputFormat.
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
[jira] [Commented] (FLINK-2167) Add fromHCat() to TableEnvironment
[
https://issues.apache.org/jira/browse/FLINK-2167?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=14933094#comment-14933094
]
ASF GitHub Bot commented on FLINK-2167:
---
Github user aljoscha commented on a diff in the pull request:
https://github.com/apache/flink/pull/1127#discussion_r40534596
--- Diff:
flink-staging/flink-table/src/main/scala/org/apache/flink/api/java/table/JavaBatchTranslator.scala
---
@@ -54,6 +56,26 @@ class JavaBatchTranslator extends PlanTranslator {
Table(Root(rowDataSet, resultFields))
}
+ override def createTable(tableSource: TableSource): Table = {
+// a TableSource requires an ExecutionEnvironment
+if (env == null) {
+ throw new InvalidProgramException("This operation requires an
ExecutionEnvironment.")
+}
+if (tableSource.isInstanceOf[AdaptiveTableSource]) {
+ val adaptiveTs = tableSource.asInstanceOf[AdaptiveTableSource]
+ if (adaptiveTs.supportsResolvedFieldPushdown ||
adaptiveTs.supportsPredicatePushdown) {
+Table(Root(adaptiveTs, adaptiveTs.getOutputFields()))
+ }
+ else {
+Table(Root(adaptiveTs.createAdaptiveDataSet(env),
adaptiveTs.getOutputFields()))
+ }
+}
+else {
+ val staticTs = tableSource.asInstanceOf[StaticTableSource]
+ createTable(staticTs.createStaticDataSet(env),
staticTs.getOutputFieldNames().mkString(","))
+}
--- End diff --
You can replace the if/else by:
```scala
tableSource match {
case adaptive: AdaptiveTableSource if
adaptive.supportsResolvedFieldPushdown && ... => ...
case adaptive: AdaptiveTableSource => ...
case static: StaticTableSource => ...
case _ => throw new Exception("something unexpected")
}
```
if I'm not mistaken, this is just of the top of my head.
> Add fromHCat() to TableEnvironment
> --
>
> Key: FLINK-2167
> URL: https://issues.apache.org/jira/browse/FLINK-2167
> Project: Flink
> Issue Type: New Feature
> Components: Table API
>Affects Versions: 0.9
>Reporter: Fabian Hueske
>Assignee: Timo Walther
>Priority: Minor
> Labels: starter
>
> Add a {{fromHCat()}} method to the {{TableEnvironment}} to read a {{Table}}
> from an HCatalog table.
> The implementation could reuse Flink's HCatInputFormat.
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
[jira] [Commented] (FLINK-2167) Add fromHCat() to TableEnvironment
[
https://issues.apache.org/jira/browse/FLINK-2167?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=14933096#comment-14933096
]
ASF GitHub Bot commented on FLINK-2167:
---
Github user aljoscha commented on a diff in the pull request:
https://github.com/apache/flink/pull/1127#discussion_r40534836
--- Diff:
flink-staging/flink-table/src/main/scala/org/apache/flink/api/java/table/JavaBatchTranslator.scala
---
@@ -118,6 +140,12 @@ class JavaBatchTranslator extends PlanTranslator {
case Root(dataSet: JavaDataSet[Row], resultFields) =>
dataSet
+ case Root(tableSource: AdaptiveTableSource, resultFields) =>
+if (env == null) {
+ throw new InvalidProgramException("This operation requires an
TableEnvironment.");
--- End diff --
semicolon
> Add fromHCat() to TableEnvironment
> --
>
> Key: FLINK-2167
> URL: https://issues.apache.org/jira/browse/FLINK-2167
> Project: Flink
> Issue Type: New Feature
> Components: Table API
>Affects Versions: 0.9
>Reporter: Fabian Hueske
>Assignee: Timo Walther
>Priority: Minor
> Labels: starter
>
> Add a {{fromHCat()}} method to the {{TableEnvironment}} to read a {{Table}}
> from an HCatalog table.
> The implementation could reuse Flink's HCatInputFormat.
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
[jira] [Commented] (FLINK-2167) Add fromHCat() to TableEnvironment
[
https://issues.apache.org/jira/browse/FLINK-2167?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=14933109#comment-14933109
]
ASF GitHub Bot commented on FLINK-2167:
---
Github user aljoscha commented on a diff in the pull request:
https://github.com/apache/flink/pull/1127#discussion_r40535960
--- Diff:
flink-staging/flink-table/src/main/scala/org/apache/flink/api/table/expressions/analysis/ResolveFieldReferences.scala
---
@@ -30,21 +33,43 @@ import org.apache.flink.api.table.trees.Rule
* Rule that resolved field references. This rule verifies that field
references point to existing
* fields of the input operation and creates [[ResolvedFieldReference]]s
that hold the field
* [[TypeInformation]] in addition to the field name.
+ *
+ * @param inputOperation is optional but required if resolved fields
should be pushed to underlying
+ * table sources.
+ * @param filtering defines if the field is resolved as a part of
filtering operation or not
*/
-class ResolveFieldReferences(inputFields: Seq[(String,
TypeInformation[_])])
+class ResolveFieldReferences(inputFields: Seq[(String,
TypeInformation[_])],
+ inputOperation: PlanNode,
+ filtering: Boolean)
--- End diff --
It is Scala Style to indent the parameters by 4 spaces (even the first
parameter) if they don't fit on a line, like so:
```scala
class ResolveFieldReferences(
inputFields: Seq[(String, TypeInformation[_])],
inputOperation: PlanNode,
filtering: Boolean) extends Rule[Expression] {
```
> Add fromHCat() to TableEnvironment
> --
>
> Key: FLINK-2167
> URL: https://issues.apache.org/jira/browse/FLINK-2167
> Project: Flink
> Issue Type: New Feature
> Components: Table API
>Affects Versions: 0.9
>Reporter: Fabian Hueske
>Assignee: Timo Walther
>Priority: Minor
> Labels: starter
>
> Add a {{fromHCat()}} method to the {{TableEnvironment}} to read a {{Table}}
> from an HCatalog table.
> The implementation could reuse Flink's HCatInputFormat.
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
[jira] [Commented] (FLINK-2167) Add fromHCat() to TableEnvironment
[
https://issues.apache.org/jira/browse/FLINK-2167?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=14933110#comment-14933110
]
ASF GitHub Bot commented on FLINK-2167:
---
Github user aljoscha commented on a diff in the pull request:
https://github.com/apache/flink/pull/1127#discussion_r40536239
--- Diff:
flink-staging/flink-table/src/main/scala/org/apache/flink/api/table/expressions/analysis/PredicatePushdown.scala
---
@@ -0,0 +1,124 @@
+/*
+ * 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.PredicateFilter.pruneExpr
+import org.apache.flink.api.table.input.{AdaptiveTableSource, TableSource}
+import org.apache.flink.api.table.plan._
+import org.apache.flink.api.table.trees.Rule
+
+import scala.collection.mutable.ArrayBuffer
+
+/**
+ * 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)._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, _) if
ts.supportsPredicatePushdown() => Seq(ts)
+case pn:PlanNode =>
+ val res = new ArrayBuffer[AdaptiveTableSource]()
+ for (child <- pn.children) res ++= getPushableTableSources(child)
+ res
--- End diff --
This can be replaced by:
```scala
pn.children flatMap { child => getPushableTableSources(child ) }
```
if I'm not mistaken, seems more Scala-y :smile:
> Add fromHCat() to TableEnvironment
> --
>
> Key: FLINK-2167
> URL: https://issues.apache.org/jira/browse/FLINK-2167
> Project: Flink
> Issue Type: New Feature
> Components: Table API
>Affects Versions: 0.9
>Reporter: Fabian Hueske
>Assignee: Timo Walther
>Priority: Minor
> Labels: starter
>
> Add a {{fromHCat()}} method to the {{TableEnvironment}} to read a {{Table}}
> from an HCatalog table.
> The implementation could reuse Flink's HCatInputFormat.
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
[jira] [Commented] (FLINK-2167) Add fromHCat() to TableEnvironment
[
https://issues.apache.org/jira/browse/FLINK-2167?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=14933117#comment-14933117
]
ASF GitHub Bot commented on FLINK-2167:
---
Github user aljoscha commented on a diff in the pull request:
https://github.com/apache/flink/pull/1127#discussion_r40536932
--- Diff:
flink-staging/flink-table/src/main/scala/org/apache/flink/api/table/expressions/analysis/PredicateFilter.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 PredicateFilter {
--- End diff --
Maybe this should be called PredicatePruner
> Add fromHCat() to TableEnvironment
> --
>
> Key: FLINK-2167
> URL: https://issues.apache.org/jira/browse/FLINK-2167
> Project: Flink
> Issue Type: New Feature
> Components: Table API
>Affects Versions: 0.9
>Reporter: Fabian Hueske
>Assignee: Timo Walther
>Priority: Minor
> Labels: starter
>
> Add a {{fromHCat()}} method to the {{TableEnvironment}} to read a {{Table}}
> from an HCatalog table.
> The implementation could reuse Flink's HCatInputFormat.
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
[jira] [Commented] (FLINK-2167) Add fromHCat() to TableEnvironment
[
https://issues.apache.org/jira/browse/FLINK-2167?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=14933128#comment-14933128
]
ASF GitHub Bot commented on FLINK-2167:
---
Github user aljoscha commented on the pull request:
https://github.com/apache/flink/pull/1127#issuecomment-143703367
Hi,
I like the work. :smile:
Some remarks about Scala style. We are trying to make our Scala code more
consistent. Mostly, I didn't comment everything, just one particular pattern of
code that could be changed.
Now, for the technical part, you are not actually removing the predicates
that are pushed down from an expression, correct? This should be alright, since
the result will still be correct. It might just be some future optimization,
but I also think that the cost of evaluating the predicate is negligible. The
real improvement comes from early filtering in the sources, as you implemented.
Then, why do you have the `supports*` methods in `AdaptiveTableSource`.
Couldn't these methods just do nothing in case the source does not support the
feature. Or maybe return false if the pushdown was not successful. (I also
wonder why you have the differentiation between AdaptiveSources that support
pushdown and those that don't in `JavaBatchTranslator.createTable`. (I think
you do it so that stuff does not get pushed to sources that don't support it,
but this distinction might not be necessary, as mentioned above.)
> Add fromHCat() to TableEnvironment
> --
>
> Key: FLINK-2167
> URL: https://issues.apache.org/jira/browse/FLINK-2167
> Project: Flink
> Issue Type: New Feature
> Components: Table API
>Affects Versions: 0.9
>Reporter: Fabian Hueske
>Assignee: Timo Walther
>Priority: Minor
> Labels: starter
>
> Add a {{fromHCat()}} method to the {{TableEnvironment}} to read a {{Table}}
> from an HCatalog table.
> The implementation could reuse Flink's HCatInputFormat.
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
[jira] [Commented] (FLINK-2167) Add fromHCat() to TableEnvironment
[
https://issues.apache.org/jira/browse/FLINK-2167?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=14933157#comment-14933157
]
ASF GitHub Bot commented on FLINK-2167:
---
Github user twalthr commented on a diff in the pull request:
https://github.com/apache/flink/pull/1127#discussion_r40540138
--- Diff: flink-staging/flink-table/pom.xml ---
@@ -37,7 +37,7 @@ under the License.
com.google.guava
guava
- ${guava.version}
+ 14.0.1
--- End diff --
Yes, I had method signature problems with the HCat input format. We are
using a very old version of HCat and therefore need a very old version of guava.
> Add fromHCat() to TableEnvironment
> --
>
> Key: FLINK-2167
> URL: https://issues.apache.org/jira/browse/FLINK-2167
> Project: Flink
> Issue Type: New Feature
> Components: Table API
>Affects Versions: 0.9
>Reporter: Fabian Hueske
>Assignee: Timo Walther
>Priority: Minor
> Labels: starter
>
> Add a {{fromHCat()}} method to the {{TableEnvironment}} to read a {{Table}}
> from an HCatalog table.
> The implementation could reuse Flink's HCatInputFormat.
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
[jira] [Commented] (FLINK-2167) Add fromHCat() to TableEnvironment
[
https://issues.apache.org/jira/browse/FLINK-2167?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=14933164#comment-14933164
]
ASF GitHub Bot commented on FLINK-2167:
---
Github user twalthr commented on a diff in the pull request:
https://github.com/apache/flink/pull/1127#discussion_r40540522
--- Diff: flink-staging/flink-table/pom.xml ---
@@ -59,6 +59,12 @@ under the License.
+ org.apache.flink
+ flink-hcatalog
+ ${project.version}
+
+
+
--- End diff --
What do you mean with "put into another package"? Do you mean putting Table
API classes in the `flink-hcatalog` package? I think thats a general decision
we have to make. Or do we want additional Maven Table API I/O format modules?
For Parquet, HCat. etc...
> Add fromHCat() to TableEnvironment
> --
>
> Key: FLINK-2167
> URL: https://issues.apache.org/jira/browse/FLINK-2167
> Project: Flink
> Issue Type: New Feature
> Components: Table API
>Affects Versions: 0.9
>Reporter: Fabian Hueske
>Assignee: Timo Walther
>Priority: Minor
> Labels: starter
>
> Add a {{fromHCat()}} method to the {{TableEnvironment}} to read a {{Table}}
> from an HCatalog table.
> The implementation could reuse Flink's HCatInputFormat.
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
[jira] [Commented] (FLINK-2167) Add fromHCat() to TableEnvironment
[
https://issues.apache.org/jira/browse/FLINK-2167?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=14933183#comment-14933183
]
ASF GitHub Bot commented on FLINK-2167:
---
Github user twalthr commented on the pull request:
https://github.com/apache/flink/pull/1127#issuecomment-143718074
Thanks @aljoscha for reviewing my code! Sorry for the bad Scala style. This
was my first really large Scala code project I wrote (not just small scripts),
I'm still learning by doing ;) I will correct the issues you mentioned.
For the technical part: Yes, I don't modify the expression tree, I'm just
giving the sources the possibility to adapt to the need of the program.
Regarding the `supports*` methods, yes you are right, actually they are not
necessary, but I thought it makes sense for possible future table sources to
check that in advance. In some cases it also reduces the amout of some method
calls, but I can also remove the `supports*` methods for reason of simplicity,
no problem.
> Add fromHCat() to TableEnvironment
> --
>
> Key: FLINK-2167
> URL: https://issues.apache.org/jira/browse/FLINK-2167
> Project: Flink
> Issue Type: New Feature
> Components: Table API
>Affects Versions: 0.9
>Reporter: Fabian Hueske
>Assignee: Timo Walther
>Priority: Minor
> Labels: starter
>
> Add a {{fromHCat()}} method to the {{TableEnvironment}} to read a {{Table}}
> from an HCatalog table.
> The implementation could reuse Flink's HCatInputFormat.
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
[jira] [Commented] (FLINK-2167) Add fromHCat() to TableEnvironment
[
https://issues.apache.org/jira/browse/FLINK-2167?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=14933224#comment-14933224
]
ASF GitHub Bot commented on FLINK-2167:
---
Github user aljoscha commented on a diff in the pull request:
https://github.com/apache/flink/pull/1127#discussion_r40545360
--- Diff: flink-staging/flink-table/pom.xml ---
@@ -59,6 +59,12 @@ under the License.
+ org.apache.flink
+ flink-hcatalog
+ ${project.version}
+
+
+
--- End diff --
I meant putting the HCat specific stuff into a different package. But I now
realize that we could not have the `fromHCat` call on `TableEnvironment` then.
So this is ok, I guess.
> Add fromHCat() to TableEnvironment
> --
>
> Key: FLINK-2167
> URL: https://issues.apache.org/jira/browse/FLINK-2167
> Project: Flink
> Issue Type: New Feature
> Components: Table API
>Affects Versions: 0.9
>Reporter: Fabian Hueske
>Assignee: Timo Walther
>Priority: Minor
> Labels: starter
>
> Add a {{fromHCat()}} method to the {{TableEnvironment}} to read a {{Table}}
> from an HCatalog table.
> The implementation could reuse Flink's HCatInputFormat.
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
[jira] [Commented] (FLINK-2167) Add fromHCat() to TableEnvironment
[
https://issues.apache.org/jira/browse/FLINK-2167?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=14933225#comment-14933225
]
ASF GitHub Bot commented on FLINK-2167:
---
Github user aljoscha commented on a diff in the pull request:
https://github.com/apache/flink/pull/1127#discussion_r40545392
--- Diff: flink-staging/flink-table/pom.xml ---
@@ -37,7 +37,7 @@ under the License.
com.google.guava
guava
- ${guava.version}
+ 14.0.1
--- End diff --
But then this could lead to problems with shading. Let's wait for @rmetzger
on this, he probably knows this stuff best.
> Add fromHCat() to TableEnvironment
> --
>
> Key: FLINK-2167
> URL: https://issues.apache.org/jira/browse/FLINK-2167
> Project: Flink
> Issue Type: New Feature
> Components: Table API
>Affects Versions: 0.9
>Reporter: Fabian Hueske
>Assignee: Timo Walther
>Priority: Minor
> Labels: starter
>
> Add a {{fromHCat()}} method to the {{TableEnvironment}} to read a {{Table}}
> from an HCatalog table.
> The implementation could reuse Flink's HCatInputFormat.
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
[jira] [Commented] (FLINK-2167) Add fromHCat() to TableEnvironment
[
https://issues.apache.org/jira/browse/FLINK-2167?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=14933227#comment-14933227
]
ASF GitHub Bot commented on FLINK-2167:
---
Github user aljoscha commented on the pull request:
https://github.com/apache/flink/pull/1127#issuecomment-143728302
Don't worry, everyone was once a starter in Scala. :smiley:
You can also leave the `supports*` methods in if you think that they might
be necessary in the future.
> Add fromHCat() to TableEnvironment
> --
>
> Key: FLINK-2167
> URL: https://issues.apache.org/jira/browse/FLINK-2167
> Project: Flink
> Issue Type: New Feature
> Components: Table API
>Affects Versions: 0.9
>Reporter: Fabian Hueske
>Assignee: Timo Walther
>Priority: Minor
> Labels: starter
>
> Add a {{fromHCat()}} method to the {{TableEnvironment}} to read a {{Table}}
> from an HCatalog table.
> The implementation could reuse Flink's HCatInputFormat.
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
[jira] [Commented] (FLINK-2167) Add fromHCat() to TableEnvironment
[
https://issues.apache.org/jira/browse/FLINK-2167?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=14935204#comment-14935204
]
ASF GitHub Bot commented on FLINK-2167:
---
Github user rmetzger commented on a diff in the pull request:
https://github.com/apache/flink/pull/1127#discussion_r40674158
--- Diff: flink-staging/flink-table/pom.xml ---
@@ -37,7 +37,7 @@ under the License.
com.google.guava
guava
- ${guava.version}
+ 14.0.1
--- End diff --
What exactly was the problem with the hcat input format and guava?
If their guava version is really incompatible with ours, we can do the
following:
- Use a newer hcat version
- create a special shaded-hcat maven module which shades away hcat's guava.
> Add fromHCat() to TableEnvironment
> --
>
> Key: FLINK-2167
> URL: https://issues.apache.org/jira/browse/FLINK-2167
> Project: Flink
> Issue Type: New Feature
> Components: Table API
>Affects Versions: 0.9
>Reporter: Fabian Hueske
>Assignee: Timo Walther
>Priority: Minor
> Labels: starter
>
> Add a {{fromHCat()}} method to the {{TableEnvironment}} to read a {{Table}}
> from an HCatalog table.
> The implementation could reuse Flink's HCatInputFormat.
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
[jira] [Commented] (FLINK-2167) Add fromHCat() to TableEnvironment
[
https://issues.apache.org/jira/browse/FLINK-2167?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=14935205#comment-14935205
]
ASF GitHub Bot commented on FLINK-2167:
---
Github user twalthr commented on a diff in the pull request:
https://github.com/apache/flink/pull/1127#discussion_r40674516
--- Diff: flink-staging/flink-table/pom.xml ---
@@ -37,7 +37,7 @@ under the License.
com.google.guava
guava
- ${guava.version}
+ 14.0.1
--- End diff --
Thanks so far, I will look into this issue again and will report.
> Add fromHCat() to TableEnvironment
> --
>
> Key: FLINK-2167
> URL: https://issues.apache.org/jira/browse/FLINK-2167
> Project: Flink
> Issue Type: New Feature
> Components: Table API
>Affects Versions: 0.9
>Reporter: Fabian Hueske
>Assignee: Timo Walther
>Priority: Minor
> Labels: starter
>
> Add a {{fromHCat()}} method to the {{TableEnvironment}} to read a {{Table}}
> from an HCatalog table.
> The implementation could reuse Flink's HCatInputFormat.
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
[jira] [Commented] (FLINK-2167) Add fromHCat() to TableEnvironment
[
https://issues.apache.org/jira/browse/FLINK-2167?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=14941064#comment-14941064
]
ASF GitHub Bot commented on FLINK-2167:
---
Github user twalthr closed the pull request at:
https://github.com/apache/flink/pull/1127
> Add fromHCat() to TableEnvironment
> --
>
> Key: FLINK-2167
> URL: https://issues.apache.org/jira/browse/FLINK-2167
> Project: Flink
> Issue Type: New Feature
> Components: Table API
>Affects Versions: 0.9
>Reporter: Fabian Hueske
>Assignee: Timo Walther
>Priority: Minor
> Labels: starter
>
> Add a {{fromHCat()}} method to the {{TableEnvironment}} to read a {{Table}}
> from an HCatalog table.
> The implementation could reuse Flink's HCatInputFormat.
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
[jira] [Commented] (FLINK-2167) Add fromHCat() to TableEnvironment
[
https://issues.apache.org/jira/browse/FLINK-2167?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=14941063#comment-14941063
]
ASF GitHub Bot commented on FLINK-2167:
---
Github user twalthr commented on the pull request:
https://github.com/apache/flink/pull/1127#issuecomment-145001030
Again, thanks for the feedback.
I will close this PR and open 2 separate PRs for TableSources and
HCatInputFormat.
> Add fromHCat() to TableEnvironment
> --
>
> Key: FLINK-2167
> URL: https://issues.apache.org/jira/browse/FLINK-2167
> Project: Flink
> Issue Type: New Feature
> Components: Table API
>Affects Versions: 0.9
>Reporter: Fabian Hueske
>Assignee: Timo Walther
>Priority: Minor
> Labels: starter
>
> Add a {{fromHCat()}} method to the {{TableEnvironment}} to read a {{Table}}
> from an HCatalog table.
> The implementation could reuse Flink's HCatInputFormat.
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
[jira] [Commented] (FLINK-2167) Add fromHCat() to TableEnvironment
[
https://issues.apache.org/jira/browse/FLINK-2167?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=14581664#comment-14581664
]
Timo Walther commented on FLINK-2167:
-
Will work on this to make myself familiar with the Table API code.
> Add fromHCat() to TableEnvironment
> --
>
> Key: FLINK-2167
> URL: https://issues.apache.org/jira/browse/FLINK-2167
> Project: Flink
> Issue Type: New Feature
> Components: Table API
>Affects Versions: 0.9
>Reporter: Fabian Hueske
>Assignee: Timo Walther
>Priority: Minor
> Labels: starter
>
> Add a {{fromHCat()}} method to the {{TableEnvironment}} to read a {{Table}}
> from an HCatalog table.
> The implementation could reuse Flink's HCatInputFormat.
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
[jira] [Commented] (FLINK-2167) Add fromHCat() to TableEnvironment
[
https://issues.apache.org/jira/browse/FLINK-2167?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=14585757#comment-14585757
]
Timo Walther commented on FLINK-2167:
-
I'm unable to connect my HCatInputFormat to the HCatalog metastore.
{code}
Configuration config = new Configuration();
config.addResource("/home/twalthr/flink/HCAT/apache-hive-1.2.0-bin/conf/hive-site.xml");
env.createInput(new HCatInputFormat("default", "mytable",
config)).print();
{code}
Is there anything I have to consider? Any hints are highly appreciated!
> Add fromHCat() to TableEnvironment
> --
>
> Key: FLINK-2167
> URL: https://issues.apache.org/jira/browse/FLINK-2167
> Project: Flink
> Issue Type: New Feature
> Components: Table API
>Affects Versions: 0.9
>Reporter: Fabian Hueske
>Assignee: Timo Walther
>Priority: Minor
> Labels: starter
>
> Add a {{fromHCat()}} method to the {{TableEnvironment}} to read a {{Table}}
> from an HCatalog table.
> The implementation could reuse Flink's HCatInputFormat.
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
[jira] [Commented] (FLINK-2167) Add fromHCat() to TableEnvironment
[
https://issues.apache.org/jira/browse/FLINK-2167?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=14594897#comment-14594897
]
Robert Metzger commented on FLINK-2167:
---
Mh. Are there any error message?
Maybe you have to load the hive-site.xml through the classloader?
> Add fromHCat() to TableEnvironment
> --
>
> Key: FLINK-2167
> URL: https://issues.apache.org/jira/browse/FLINK-2167
> Project: Flink
> Issue Type: New Feature
> Components: Table API
>Affects Versions: 0.9
>Reporter: Fabian Hueske
>Assignee: Timo Walther
>Priority: Minor
> Labels: starter
>
> Add a {{fromHCat()}} method to the {{TableEnvironment}} to read a {{Table}}
> from an HCatalog table.
> The implementation could reuse Flink's HCatInputFormat.
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
[jira] [Commented] (FLINK-2167) Add fromHCat() to TableEnvironment
[
https://issues.apache.org/jira/browse/FLINK-2167?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=14599279#comment-14599279
]
Timo Walther commented on FLINK-2167:
-
Thanks Robert, I solved the issue. The hive-site.xml has to be in the
classpath. Everything else does not work.
[~aljoscha] How should Scala users access "fromHCat()"? Shall I create a
separate TableEnviroment for the Scala API? Should the ExecutionEnvironment be
converted to a TableEnvironment implicitly?
In general: How should I create tests? Adding Hive as a dependency could be a
little bit too much...
> Add fromHCat() to TableEnvironment
> --
>
> Key: FLINK-2167
> URL: https://issues.apache.org/jira/browse/FLINK-2167
> Project: Flink
> Issue Type: New Feature
> Components: Table API
>Affects Versions: 0.9
>Reporter: Fabian Hueske
>Assignee: Timo Walther
>Priority: Minor
> Labels: starter
>
> Add a {{fromHCat()}} method to the {{TableEnvironment}} to read a {{Table}}
> from an HCatalog table.
> The implementation could reuse Flink's HCatInputFormat.
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
[jira] [Commented] (FLINK-2167) Add fromHCat() to TableEnvironment
[
https://issues.apache.org/jira/browse/FLINK-2167?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=14599368#comment-14599368
]
Aljoscha Krettek commented on FLINK-2167:
-
Does it not work we simply use the same TableEnvironment? What is the result of
the {{fromHCat()}} call?
> Add fromHCat() to TableEnvironment
> --
>
> Key: FLINK-2167
> URL: https://issues.apache.org/jira/browse/FLINK-2167
> Project: Flink
> Issue Type: New Feature
> Components: Table API
>Affects Versions: 0.9
>Reporter: Fabian Hueske
>Assignee: Timo Walther
>Priority: Minor
> Labels: starter
>
> Add a {{fromHCat()}} method to the {{TableEnvironment}} to read a {{Table}}
> from an HCatalog table.
> The implementation could reuse Flink's HCatInputFormat.
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
[jira] [Commented] (FLINK-2167) Add fromHCat() to TableEnvironment
[
https://issues.apache.org/jira/browse/FLINK-2167?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=14599422#comment-14599422
]
Timo Walther commented on FLINK-2167:
-
Sure that does also work. I just thought it is confusing if the user uses a
TableEnvironment that is located in a "XXX.java.XXX" package for a Scala
program.
The result would be a Table.
> Add fromHCat() to TableEnvironment
> --
>
> Key: FLINK-2167
> URL: https://issues.apache.org/jira/browse/FLINK-2167
> Project: Flink
> Issue Type: New Feature
> Components: Table API
>Affects Versions: 0.9
>Reporter: Fabian Hueske
>Assignee: Timo Walther
>Priority: Minor
> Labels: starter
>
> Add a {{fromHCat()}} method to the {{TableEnvironment}} to read a {{Table}}
> from an HCatalog table.
> The implementation could reuse Flink's HCatInputFormat.
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
[jira] [Commented] (FLINK-2167) Add fromHCat() to TableEnvironment
[
https://issues.apache.org/jira/browse/FLINK-2167?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=14599474#comment-14599474
]
Aljoscha Krettek commented on FLINK-2167:
-
Ok, maybe if we move it to another package?
> Add fromHCat() to TableEnvironment
> --
>
> Key: FLINK-2167
> URL: https://issues.apache.org/jira/browse/FLINK-2167
> Project: Flink
> Issue Type: New Feature
> Components: Table API
>Affects Versions: 0.9
>Reporter: Fabian Hueske
>Assignee: Timo Walther
>Priority: Minor
> Labels: starter
>
> Add a {{fromHCat()}} method to the {{TableEnvironment}} to read a {{Table}}
> from an HCatalog table.
> The implementation could reuse Flink's HCatInputFormat.
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
[jira] [Commented] (FLINK-2167) Add fromHCat() to TableEnvironment
[
https://issues.apache.org/jira/browse/FLINK-2167?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=14599502#comment-14599502
]
Timo Walther commented on FLINK-2167:
-
Moving is also an option. I think I will implement what I think is best and
then we can discuss afterwards. Because there will be many changes necessary.
E.g. I think it makes sense to pass an the {{ExecutionEnvironment}} to the
{{TableEnvironment}}, otherwise I can't figure out if we are in the DataStream
or DataSet area for the input format.
> Add fromHCat() to TableEnvironment
> --
>
> Key: FLINK-2167
> URL: https://issues.apache.org/jira/browse/FLINK-2167
> Project: Flink
> Issue Type: New Feature
> Components: Table API
>Affects Versions: 0.9
>Reporter: Fabian Hueske
>Assignee: Timo Walther
>Priority: Minor
> Labels: starter
>
> Add a {{fromHCat()}} method to the {{TableEnvironment}} to read a {{Table}}
> from an HCatalog table.
> The implementation could reuse Flink's HCatInputFormat.
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
[jira] [Commented] (FLINK-2167) Add fromHCat() to TableEnvironment
[
https://issues.apache.org/jira/browse/FLINK-2167?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=14599630#comment-14599630
]
Aljoscha Krettek commented on FLINK-2167:
-
Yes, this seems necessary.
> Add fromHCat() to TableEnvironment
> --
>
> Key: FLINK-2167
> URL: https://issues.apache.org/jira/browse/FLINK-2167
> Project: Flink
> Issue Type: New Feature
> Components: Table API
>Affects Versions: 0.9
>Reporter: Fabian Hueske
>Assignee: Timo Walther
>Priority: Minor
> Labels: starter
>
> Add a {{fromHCat()}} method to the {{TableEnvironment}} to read a {{Table}}
> from an HCatalog table.
> The implementation could reuse Flink's HCatInputFormat.
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
