Skip to content

[FLINK-22454][table-planner-blink] only ignore casting on interoperable type when extracting lookup keys #16094

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 1 commit into from
Jun 10, 2021
Merged

[FLINK-22454][table-planner-blink] only ignore casting on interoperable type when extracting lookup keys #16094

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 6 additions & 1 deletion .../org/apache/flink/table/planner/plan/nodes/physical/common/CommonPhysicalLookupJoin.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,7 @@ import org.apache.flink.table.planner.plan.utils.LookupJoinUtil._
import org.apache.flink.table.planner.plan.utils.PythonUtil.containsPythonCall
import org.apache.flink.table.planner.plan.utils.RelExplainUtil.preferExpressionFormat
import org.apache.flink.table.planner.plan.utils.{JoinTypeUtil, LookupJoinUtil, RelExplainUtil}
import org.apache.flink.table.runtime.types.PlannerTypeUtils

import org.apache.calcite.plan.{RelOptCluster, RelOptTable, RelTraitSet}
import org.apache.calcite.rel.`type`.{RelDataType, RelDataTypeField}
Expand Down Expand Up @@ -295,7 +296,11 @@ abstract class CommonPhysicalLookupJoin(
expr = call.getOperands.get(0)
case call: RexCall if call.getOperator == SqlStdOperatorTable.CAST =>
// drill through identity function
expr = call.getOperands.get(0)
val outputType = call.getType
val inputType = call.getOperands.get(0).getType
val isCompatible = PlannerTypeUtils.isInteroperable(
FlinkTypeFactory.toLogicalType(outputType), FlinkTypeFactory.toLogicalType(inputType))
expr = if (isCompatible) call.getOperands.get(0) else expr
case _ =>
}
expr match {
Expand Down
48 changes: 48 additions & 0 deletions ...k/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/join/LookupJoinTest.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -474,6 +474,54 @@ class LookupJoinTest(legacyTableSource: Boolean) extends TableTestBase with Seri
util.verifyExecPlan(sql)
}

@Test
def testJoinTemporalTableWithCastOnLookupTable(): Unit = {
util.addTable(
"""
|CREATE TABLE LookupTable2 (
| `id` decimal(38, 18),
| `name` STRING,
| `age` INT
|) WITH (
| 'connector' = 'values'
|)
|""".stripMargin)
val sql =
"""
|SELECT MyTable.b, LookupTable2.id
|FROM MyTable
|LEFT JOIN LookupTable2 FOR SYSTEM_TIME AS OF MyTable.`proctime`
|ON MyTable.a = CAST(LookupTable2.`id` as INT)
|""".stripMargin
thrown.expect(classOf[TableException])
thrown.expectMessage("Temporal table join requires an equality condition on fields of " +
"table [default_catalog.default_database.LookupTable2]")
verifyTranslationSuccess(sql)
}

@Test
def testJoinTemporalTableWithInteroperableCastOnLookupTable(): Unit = {
util.addTable(
"""
|CREATE TABLE LookupTable2 (
| `id` INT,
| `name` char(10),
| `age` INT
|) WITH (
| 'connector' = 'values'
|)
|""".stripMargin)

val sql =
"""
|SELECT MyTable.b, LookupTable2.id
|FROM MyTable
|LEFT JOIN LookupTable2 FOR SYSTEM_TIME AS OF MyTable.`proctime`
|ON MyTable.b = CAST(LookupTable2.`name` as String)
|""".stripMargin
verifyTranslationSuccess(sql)
}

// ==========================================================================================

private def createLookupTable(tableName: String, lookupFunction: UserDefinedFunction): Unit = {
Expand Down