This is an automated email from the ASF dual-hosted git repository.
gustavodemorais pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new cce8840bbfc [FLINK-39125][table] Support injective casts from
BINARY/VARBINARY to CHAR/VARCHAR for upsert keys
cce8840bbfc is described below
commit cce8840bbfc8825ea76510855c4323679f68ed8d
Author: Gustavo de Morais <[email protected]>
AuthorDate: Mon Jun 22 11:27:56 2026 +0200
[FLINK-39125][table] Support injective casts from BINARY/VARBINARY to
CHAR/VARCHAR for upsert keys
This closes #28480.
---
.../types/logical/utils/LogicalTypeCasts.java | 25 ++++++++++
.../flink/table/types/LogicalTypeCastsTest.java | 33 +++++++++++--
.../plan/metadata/FlinkRelMdUniqueKeys.scala | 36 +++++++++-----
.../plan/metadata/FlinkRelMdUpsertKeys.scala | 5 +-
.../plan/metadata/FlinkRelMdUpsertKeysTest.scala | 57 ++++++++++++++++++++--
5 files changed, 136 insertions(+), 20 deletions(-)
diff --git
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/logical/utils/LogicalTypeCasts.java
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/logical/utils/LogicalTypeCasts.java
index 85de267a368..a0afca2caa0 100644
---
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/logical/utils/LogicalTypeCasts.java
+++
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/logical/utils/LogicalTypeCasts.java
@@ -167,6 +167,29 @@ public final class LogicalTypeCasts {
return (long) getLength(target) >= (long) getLength(source) *
4;
};
+ /**
+ * Injective when the target char length can hold a UTF-8 byte string (at
most one char per
+ * byte).
+ */
+ private static final BiPredicate<LogicalType, LogicalType>
WHEN_CHAR_LENGTH_FITS_UTF8 =
+ (source, target) -> {
+ // Only CHAR with max length is safe.
+ // Bounded CHAR right-pads short values with spaces, so
distinct inputs collide.
+ // Example of collision for CHAR(3):
+ // bytes [0x61] -> "a" -> "a "
+ // bytes [0x61,0x20,0x20] -> "a "
+ if (target.is(CHAR) && !hasMaxLength(target)) {
+ return false;
+ }
+ if (hasMaxLength(target)) {
+ return true;
+ }
+ if (hasMaxLength(source)) {
+ return false;
+ }
+ return getLength(target) >= getLength(source);
+ };
+
static {
implicitCastingRules = new HashMap<>();
explicitCastingRules = new HashMap<>();
@@ -188,6 +211,7 @@ public final class LogicalTypeCasts {
.explicitFrom(RAW, NULL, STRUCTURED_TYPE, BITMAP)
.injectiveFrom(WHEN_LENGTH_FITS, CHAR)
.injectiveFrom(WHEN_MAX_CHAR_LENGTH_FITS,
STRING_INJECTIVE_SOURCES)
+ .injectiveFrom(WHEN_CHAR_LENGTH_FITS_UTF8, BINARY, VARBINARY)
.build();
castTo(VARCHAR)
@@ -196,6 +220,7 @@ public final class LogicalTypeCasts {
.explicitFrom(RAW, NULL, STRUCTURED_TYPE, BITMAP)
.injectiveFrom(WHEN_LENGTH_FITS, CHAR, VARCHAR)
.injectiveFrom(WHEN_MAX_CHAR_LENGTH_FITS,
STRING_INJECTIVE_SOURCES)
+ .injectiveFrom(WHEN_CHAR_LENGTH_FITS_UTF8, BINARY, VARBINARY)
.build();
//
-----------------------------------------------------------------------------------------
diff --git
a/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/LogicalTypeCastsTest.java
b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/LogicalTypeCastsTest.java
index e6ac53f79bd..3d0dec23997 100644
---
a/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/LogicalTypeCastsTest.java
+++
b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/LogicalTypeCastsTest.java
@@ -340,9 +340,9 @@ class LogicalTypeCastsTest {
// DECIMAL to STRING is NOT considered injective
Arguments.of(new DecimalType(10, 2), VarCharType.STRING_TYPE,
false),
- // BYTES to STRING is NOT injective (invalid UTF-8 sequences
collapse)
- Arguments.of(new VarBinaryType(100), VarCharType.STRING_TYPE,
false),
- Arguments.of(new BinaryType(100), VarCharType.STRING_TYPE,
false),
+ // BYTES to STRING is injective: UTF-8 decodes to at most one
char per byte
+ Arguments.of(new VarBinaryType(100), VarCharType.STRING_TYPE,
true),
+ Arguments.of(new BinaryType(100), VarCharType.STRING_TYPE,
true),
// TIMESTAMP_WITH_TIME_ZONE to STRING is NOT injective
// (theory: two timestamps with different zones could produce
same string
@@ -507,7 +507,32 @@ class LogicalTypeCastsTest {
// CHAR(10) → VARBINARY(40): fixed char to var binary
Arguments.of(new CharType(10), new VarBinaryType(40), true),
// VARCHAR(10) → BINARY(40): var char to fixed binary
- Arguments.of(new VarCharType(10), new BinaryType(40), true));
+ Arguments.of(new VarCharType(10), new BinaryType(40), true),
+
+ // ---- Binary to string injective casts (UTF-8: at most one
char per byte) ----
+
+ // VARBINARY(MAX) → VARCHAR(MAX): both unbounded
+ Arguments.of(
+ new VarBinaryType(VarBinaryType.MAX_LENGTH),
VarCharType.STRING_TYPE, true),
+ // BINARY(MAX) → VARCHAR(MAX): both unbounded
+ Arguments.of(new BinaryType(BinaryType.MAX_LENGTH),
VarCharType.STRING_TYPE, true),
+ // VARBINARY(10) → VARCHAR(10): exact fit
+ Arguments.of(new VarBinaryType(10), new VarCharType(10), true),
+ // VARBINARY(10) → VARCHAR(9): one char short
+ Arguments.of(new VarBinaryType(10), new VarCharType(9), false),
+ // VARBINARY(10) → VARCHAR(MAX): bounded source, unbounded
target
+ Arguments.of(new VarBinaryType(10), VarCharType.STRING_TYPE,
true),
+ // VARBINARY(MAX) → VARCHAR(100): unbounded source, bounded
target
+ Arguments.of(
+ new VarBinaryType(VarBinaryType.MAX_LENGTH), new
VarCharType(100), false),
+ // BINARY(10) → VARCHAR(10): fixed binary to var char, exact
fit
+ Arguments.of(new BinaryType(10), new VarCharType(10), true),
+ // VARBINARY(10) → CHAR(10): bounded CHAR pads short values,
NOT injective
+ Arguments.of(new VarBinaryType(10), new CharType(10), false),
+ // VARBINARY(10) → CHAR(MAX): unbounded CHAR does not pad,
injective
+ Arguments.of(new VarBinaryType(10), new
CharType(CharType.MAX_LENGTH), true),
+ // BINARY(10) → CHAR(20): bounded CHAR pads even when wider,
NOT injective
+ Arguments.of(new BinaryType(10), new CharType(20), false));
}
@ParameterizedTest(name = "{index}: [From: {0}, To: {1}, Injective: {2}]")
diff --git
a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/metadata/FlinkRelMdUniqueKeys.scala
b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/metadata/FlinkRelMdUniqueKeys.scala
index c445d8b2949..da15806cb98 100644
---
a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/metadata/FlinkRelMdUniqueKeys.scala
+++
b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/metadata/FlinkRelMdUniqueKeys.scala
@@ -17,6 +17,7 @@
*/
package org.apache.flink.table.planner.plan.metadata
+import org.apache.flink.table.api.config.ExecutionConfigOptions
import org.apache.flink.table.catalog.{CatalogTable, ResolvedCatalogBaseTable}
import org.apache.flink.table.connector.ChangelogMode
import org.apache.flink.table.planner._
@@ -27,8 +28,10 @@ import
org.apache.flink.table.planner.plan.nodes.physical.common.CommonPhysicalL
import org.apache.flink.table.planner.plan.nodes.physical.stream._
import org.apache.flink.table.planner.plan.schema.{FlinkPreparingTableBase,
TableSourceTable}
import org.apache.flink.table.planner.plan.utils.{ChangelogPlanUtils,
FlinkRelMdUtil, RankUtil}
+import org.apache.flink.table.planner.utils.ShortcutUtils.unwrapTableConfig
import org.apache.flink.table.runtime.groupwindow.NamedWindowProperty
import org.apache.flink.table.runtime.operators.rank.RankType
+import org.apache.flink.table.types.logical.LogicalTypeFamily
import org.apache.flink.table.types.logical.utils.LogicalTypeCasts
import org.apache.flink.types.RowKind
@@ -36,7 +39,6 @@ import com.google.common.collect.ImmutableSet
import org.apache.calcite.plan.RelOptTable
import org.apache.calcite.plan.hep.HepRelVertex
import org.apache.calcite.plan.volcano.RelSubset
-import org.apache.calcite.rel.`type`.RelDataTypeFactory
import org.apache.calcite.rel.{RelNode, SingleRel}
import org.apache.calcite.rel.core._
import org.apache.calcite.rel.metadata._
@@ -126,19 +128,15 @@ class FlinkRelMdUniqueKeys private extends
MetadataHandler[BuiltInMetadata.Uniqu
projects: JList[RexNode],
input: RelNode,
mq: RelMetadataQuery,
- ignoreNulls: Boolean): JSet[ImmutableBitSet] = {
- getProjectUniqueKeys(
- projects,
- input.getCluster.getTypeFactory,
- () => mq.getUniqueKeys(input, ignoreNulls),
- ignoreNulls)
- }
+ ignoreNulls: Boolean): JSet[ImmutableBitSet] =
+ getProjectUniqueKeys(projects, input, () => mq.getUniqueKeys(input,
ignoreNulls), ignoreNulls)
def getProjectUniqueKeys(
projects: JList[RexNode],
- typeFactory: RelDataTypeFactory,
+ input: RelNode,
getInputUniqueKeys: () => util.Set[ImmutableBitSet],
ignoreNulls: Boolean): JSet[ImmutableBitSet] = {
+ val typeFactory = input.getCluster.getTypeFactory
// LogicalProject maps a set of rows to a different set;
// Without knowledge of the mapping function(whether it
// preserves uniqueness), it is only safe to derive uniqueness
@@ -175,7 +173,8 @@ class FlinkRelMdUniqueKeys private extends
MetadataHandler[BuiltInMetadata.Uniqu
}
// rename or key-preserving cast (fidelity or injective)
case a: RexCall
- if (a.getKind.equals(SqlKind.AS) || isKeyPreservingCast(a)) &&
+ if (a.getKind.equals(SqlKind.AS) ||
+ isKeyPreservingCast(a, legacyBytesToStringCast(input))) &&
a.getOperands.get(0).isInstanceOf[RexInputRef] =>
appendMapInToOutPos(a.getOperands.get(0).asInstanceOf[RexInputRef].getIndex, i)
case _ => // ignore
@@ -219,16 +218,31 @@ class FlinkRelMdUniqueKeys private extends
MetadataHandler[BuiltInMetadata.Uniqu
*
* An injective cast is one where each distinct input maps to a distinct
output, ensuring that
* unique keys remain unique after the cast.
+ *
+ * In legacy bytes-to-string mode, invalid UTF-8 is replaced by U+FFFD, so
distinct byte arrays
+ * can map to the same string. A BINARY/VARBINARY -> CHAR/VARCHAR cast is
therefore not injective
+ * under that mode.
*/
- private def isKeyPreservingCast(call: RexCall): Boolean = {
+ private def isKeyPreservingCast(call: RexCall, legacyBytesToStringCast:
Boolean): Boolean = {
if (call.getKind != SqlKind.CAST) {
return false
}
val originalType =
FlinkTypeFactory.toLogicalType(call.getOperands.get(0).getType)
val newType = FlinkTypeFactory.toLogicalType(call.getType)
+ if (
+ legacyBytesToStringCast &&
originalType.is(LogicalTypeFamily.BINARY_STRING) &&
+ newType.is(LogicalTypeFamily.CHARACTER_STRING)
+ ) {
+ return false
+ }
LogicalTypeCasts.supportsInjectiveCast(originalType, newType)
}
+ // Can be removed together with the legacyBytesToStringCast parameter once
+ // TABLE_EXEC_LEGACY_BYTES_TO_STRING_CAST is dropped.
+ private def legacyBytesToStringCast(input: RelNode): Boolean =
+
unwrapTableConfig(input).get(ExecutionConfigOptions.TABLE_EXEC_LEGACY_BYTES_TO_STRING_CAST)
+
def getUniqueKeys(
rel: Expand,
mq: RelMetadataQuery,
diff --git
a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/metadata/FlinkRelMdUpsertKeys.scala
b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/metadata/FlinkRelMdUpsertKeys.scala
index 7890b116b26..702359d3ee5 100644
---
a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/metadata/FlinkRelMdUpsertKeys.scala
+++
b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/metadata/FlinkRelMdUpsertKeys.scala
@@ -74,9 +74,10 @@ class FlinkRelMdUpsertKeys private extends
MetadataHandler[UpsertKeys] {
mq: RelMetadataQuery): JSet[ImmutableBitSet] =
FlinkRelMdUniqueKeys.INSTANCE.getProjectUniqueKeys(
projects,
- input.getCluster.getTypeFactory,
+ input,
() => FlinkRelMetadataQuery.reuseOrCreate(mq).getUpsertKeys(input),
- ignoreNulls = false)
+ ignoreNulls = false
+ )
def getUpsertKeys(rel: Expand, mq: RelMetadataQuery): JSet[ImmutableBitSet] =
FlinkRelMdUniqueKeys.INSTANCE.getExpandUniqueKeys(
diff --git
a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/metadata/FlinkRelMdUpsertKeysTest.scala
b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/metadata/FlinkRelMdUpsertKeysTest.scala
index 4ae6f389e51..3fdb049dab8 100644
---
a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/metadata/FlinkRelMdUpsertKeysTest.scala
+++
b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/metadata/FlinkRelMdUpsertKeysTest.scala
@@ -17,6 +17,7 @@
*/
package org.apache.flink.table.planner.plan.metadata
+import org.apache.flink.table.api.config.ExecutionConfigOptions
import org.apache.flink.table.planner.plan.`trait`.FlinkRelDistribution
import org.apache.flink.table.planner.plan.nodes.calcite.LogicalExpand
import
org.apache.flink.table.planner.plan.nodes.logical.FlinkLogicalOverAggregate
@@ -27,12 +28,13 @@ import
org.apache.flink.table.runtime.operators.rank.{ConstantRankRange, RankTyp
import com.google.common.collect.{ImmutableList, ImmutableSet}
import org.apache.calcite.prepare.CalciteCatalogReader
-import org.apache.calcite.rel.`type`.RelDataTypeFieldImpl
+import org.apache.calcite.rel.`type`.{RelDataType, RelDataTypeFieldImpl}
import org.apache.calcite.rel.{RelCollations, RelFieldCollation, RelNode}
import org.apache.calcite.rel.core.{JoinRelType, Window}
import org.apache.calcite.rel.hint.RelHint
+import org.apache.calcite.rel.logical.LogicalProject
import org.apache.calcite.rex.{RexInputRef, RexNode, RexWindowBounds}
-import org.apache.calcite.sql.`type`.SqlTypeName.VARCHAR
+import org.apache.calcite.sql.`type`.SqlTypeName.{VARBINARY, VARCHAR}
import org.apache.calcite.sql.SqlWindow
import org.apache.calcite.sql.fun.SqlStdOperatorTable.{EQUALS, LESS_THAN, MAX}
import org.apache.calcite.sql.parser.SqlParserPos
@@ -42,7 +44,7 @@ import org.junit.jupiter.api.Test
import java.util.Collections
-import scala.collection.JavaConverters.setAsJavaSetConverter
+import scala.collection.JavaConverters.{seqAsJavaListConverter,
setAsJavaSetConverter}
class FlinkRelMdUpsertKeysTest extends FlinkRelMdHandlerTestBase {
@@ -174,6 +176,55 @@ class FlinkRelMdUpsertKeysTest extends
FlinkRelMdHandlerTestBase {
assertEquals(toBitSet(), mq.getUpsertKeys(narrowedProject))
}
+ @Test
+ def testGetUpsertKeysOnProjectInjectiveBytesToStringCast(): Unit = {
+ // VARCHAR(100) -> VARBINARY(400) is injective: 400 bytes covers the
worst-case UTF-8
+ // expansion (100 chars * 4 bytes), so the key survives.
+ val keyedByString = castField(studentLogicalScan, 0,
typeFactory.createSqlType(VARCHAR, 100))
+ val keyedByBytes = castField(keyedByString, 0,
typeFactory.createSqlType(VARBINARY, 400))
+ assertEquals(toBitSet(Array(0)), mq.getUpsertKeys(keyedByBytes))
+
+ // VARBINARY(400) -> VARCHAR(400) is injective, so the key survives.
+ val keyedByStringAgain = castField(keyedByBytes, 0,
typeFactory.createSqlType(VARCHAR, 400))
+ assertEquals(toBitSet(Array(0)), mq.getUpsertKeys(keyedByStringAgain))
+ }
+
+ @Test
+ def testGetUpsertKeysOnProjectBytesToStringCastLosesKeyInLegacyMode(): Unit
= {
+ // Legacy mode replaces invalid UTF-8 with U+FFFD, so distinct bytes can
collide and the cast
+ // is no longer injective.
+ tableConfig.set(
+ ExecutionConfigOptions.TABLE_EXEC_LEGACY_BYTES_TO_STRING_CAST,
+ Boolean.box(true))
+
+ val keyedByString = castField(studentLogicalScan, 0,
typeFactory.createSqlType(VARCHAR, 100))
+ val keyedByBytes = castField(keyedByString, 0,
typeFactory.createSqlType(VARBINARY, 400))
+ val keyedByStringAgain = castField(keyedByBytes, 0,
typeFactory.createSqlType(VARCHAR, 400))
+
+ assertEquals(toBitSet(), mq.getUpsertKeys(keyedByStringAgain))
+ }
+
+ /**
+ * Projects `input` with the field at `fieldIndex` cast to `targetType` and
all other fields
+ * passed through unchanged, preserving the input field names.
+ */
+ private def castField(
+ input: RelNode,
+ fieldIndex: Int,
+ targetType: RelDataType): LogicalProject = {
+ val rowType = input.getRowType
+ val exprs: Seq[RexNode] = (0 until rowType.getFieldCount).map {
+ i =>
+ val ref = RexInputRef.of(i, rowType)
+ if (i == fieldIndex) rexBuilder.makeCast(targetType, ref) else ref
+ }
+ LogicalProject.create(
+ input,
+ Collections.emptyList[RelHint](),
+ exprs.asJava,
+ rowType.getFieldNames)
+ }
+
@Test
def testGetUpsertKeysOnFilter(): Unit = {
assertEquals(toBitSet(Array(0)), mq.getUpsertKeys(logicalFilter))
