[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)