[FLINK-4565] [table] Support for SQL IN operator #4404

twalthr · 2017-07-26T17:26:37Z

What is the purpose of the change

This PR adds support for the IN operator in both SQL and Table API. Both for testing in a set of elements f0.in(1, 2, 3) as well as in sub-query f0.in(table)

Brief change log

Various changes in plan translation, code generation, and API.

Verifying this change

This change added tests and can be verified as follows:

Unit tests: org.apache.flink.table.expressions.ScalarOperatorsTest
Unit tests: org.apache.flink.table.expressions.validation.ScalarOperatorsValidationTest
Integration tests: org.apache.flink.table.runtime.batch.sql.SetOperatorsITCase
Integration tests: org.apache.flink.table.runtime.batch.table.SetOperatorsITCase

Does this pull request potentially affect one of the following parts:

Dependencies (does it add or upgrade a dependency): no
The public API, i.e., is any changed class annotated with @Public(Evolving): no
The serializers: no
The runtime per-record code paths (performance sensitive): no
Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Yarn/Mesos, ZooKeeper: no

Documentation

Does this pull request introduce a new feature? yes
If yes, how is the feature documented? not documented

fhueske

Hi @twalthr, the change itself looks very good. I had just one minor comment about TableReference.

I would change the tests from runtime tests to plan tests though. The change does not add a new operator, just expressions (which are separately tested) and translation to regular (tested) joins. I think we also need tests to validate the equivalence of expression and string Table API.

It would be good to extend the documentation as well.

What do you think?
Fabian

fhueske · 2017-08-01T20:08:05Z

...ries/flink-table/src/test/scala/org/apache/flink/table/expressions/ScalarOperatorsTest.scala

@@ -137,6 +133,72 @@ class ScalarOperatorsTest extends ScalarOperatorsTestBase {
  }

  @Test
+  def testIn(): Unit = {


Also add tests for expressions that result in false?

fhueske · 2017-08-02T11:51:10Z

...ibraries/flink-table/src/main/scala/org/apache/flink/table/expressions/fieldExpression.scala

@@ -138,6 +138,25 @@ case class WindowReference(name: String, tpe: Option[TypeInformation[_]] = None)
  override def toString: String = s"'$name"
 }

+case class TableReference(name: String, table: Table) extends Attribute {


TableReference shouldn't extend Attribute IMO.
We can make it extend LeafExpression with NamedExpression.

fhueske · 2017-08-02T12:13:40Z

...ink-table/src/test/scala/org/apache/flink/table/runtime/batch/table/SetOperatorsITCase.scala

@@ -225,4 +227,62 @@ class SetOperatorsITCase(
    val expected = "2,1,Hi\n" + "3,2,Hello\n" + "4,2,Hello world\n"
    TestBaseUtils.compareResultAsText(results.asJava, expected)
  }
+
+  @Test
+  def testInWithFilter(): Unit = {


Add stringexpr tests to validate equivalence of Scala expression and Java String Table API.

These runtime tests could also be replaced by plan tests, IMO.

fhueske · 2017-08-02T12:17:02Z

...flink-table/src/test/scala/org/apache/flink/table/runtime/batch/sql/SetOperatorsITCase.scala

@@ -260,4 +260,108 @@ class SetOperatorsITCase(
    val results = result.toDataSet[Row].collect()
    TestBaseUtils.compareResultAsText(results.asJava, expected)
  }
+
+  @Test
+  def testInWithFilter(): Unit = {


Should we validate the feature with plan tests (TableTestBase) rather than runtime tests?
The feature does not require new operators, just expressions which are tested by dedicated tests and joins which are also tested.

fhueske · 2017-08-02T12:23:34Z

flink-tests/src/test/scala/org/apache/flink/api/scala/util/CollectionDataSets.scala

@@ -17,9 +17,9 @@
 */
 package org.apache.flink.api.scala.util

+import java.sql.{Date, Time, Timestamp}


remove unused imports

fhueske · 2017-08-02T12:24:01Z

...ibraries/flink-table/src/test/scala/org/apache/flink/table/runtime/utils/BatchTestData.scala

+import scala.collection.mutable
+import scala.util.Random
+
+object BatchTestData {


Not needed if we switch to plan tests.

…eries

twalthr · 2017-08-04T08:42:26Z

Thanks for reviewing @fhueske. I addressed your feedback and added documentation. I will merge this now...

tedyu · 2017-08-05T01:40:07Z

docs/dev/table/sql.md

+        <span class="label label-primary">Batch</span>
+      </td>
+      <td>
+      Returns true if an expression exists in a given table sub-query. The sub-query table must consist of one column. This column must have the same data type as the expression.


Should the syntax be enhanced so that user can specify one column in the table with multiple columns ?

What do you mean with "enhanced"? We cannot change the SQL standard. The user can nest another query and reduce the table to one column.

tedyu · 2017-08-05T10:02:23Z

flink-libraries/flink-table/src/main/scala/org/apache/flink/table/expressions/subquery.scala

+        }
+        (expression.resultType, tableOutput.head.resultType) match {
+          case (lType, rType) if isNumeric(lType) && isNumeric(rType) => ValidationSuccess
+          case (lType, rType) if lType == rType => ValidationSuccess


Should this be moved ahead of the isNumeric() check since this check is cheaper ?

Good point, yes we could do it. Feel free to submit a patch for it.

Should I open a new JIRA ?

I think a hotfix is enough. This part is not performance critical anyway, because it happens before runtime.

Created #4493

@twalthr:
Can you take a look at my PR ?

tedyu · 2017-08-09T08:34:36Z

@fhueske
Can you take a look at #4493 ?

fhueske reviewed Aug 2, 2017

View reviewed changes

DmytroShkvyra and others added 3 commits August 3, 2017 12:03

[FLINK-4565] [table] Support for SQL IN operator

8810fcf

[FLINK-4565] [table] Full IN operator support for literals and sub-qu…

fd46bc9

…eries

Feedback addressed

758a5cc

twalthr force-pushed the FLINK-4565 branch from 14ff3c2 to 758a5cc Compare August 4, 2017 07:47

Documentation added

821b046

asfgit closed this in 4f776a2 Aug 4, 2017

tedyu reviewed Aug 5, 2017

View reviewed changes

rmetzger added the component=API/TableSQL label Mar 14, 2019

flinkbot added component=TableSQL/API and removed component=API/TableSQL labels Mar 17, 2022

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[FLINK-4565] [table] Support for SQL IN operator #4404

[FLINK-4565] [table] Support for SQL IN operator #4404

twalthr commented Jul 26, 2017

fhueske left a comment

fhueske Aug 1, 2017

fhueske Aug 2, 2017

fhueske Aug 2, 2017

fhueske Aug 2, 2017

fhueske Aug 2, 2017

fhueske Aug 2, 2017

fhueske Aug 2, 2017

twalthr commented Aug 4, 2017

tedyu Aug 5, 2017

twalthr Aug 7, 2017

tedyu Aug 5, 2017

twalthr Aug 7, 2017

tedyu Aug 7, 2017

twalthr Aug 7, 2017

tedyu Aug 7, 2017

tedyu Aug 8, 2017

tedyu commented Aug 9, 2017

[FLINK-4565] [table] Support for SQL IN operator #4404

[FLINK-4565] [table] Support for SQL IN operator #4404

Conversation

twalthr commented Jul 26, 2017

What is the purpose of the change

Brief change log

Verifying this change

Does this pull request potentially affect one of the following parts:

Documentation

fhueske left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

twalthr commented Aug 4, 2017

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

tedyu commented Aug 9, 2017