[FLINK-22901] Introduce getChangeLogUpsertKeys in FlinkRelMetadataQuery - ASF JIRA

Attach files

Attach Screenshot

Voters

Watch issue

Watchers

Link

Clone

Update Comment Author

Replace String in Comment

Update Comment Visibility

Delete Comments

XML

Word

Printable

JSON

Details

Type: Sub-task
Status: Closed
Priority: Major
Resolution: Fixed
Affects Version/s: None
Fix Version/s: 1.14.0, 1.13.3
Component/s: Table SQL / Planner
Labels:
- pull-request-available

Description

For fix ~~FLINK-20374~~, we need to resolve streaming computation disorder. we need to introduce a change log upsert keys, this is not unique keys.

/**
 * Determines the set of change log upsert minimal keys for this expression. A key is
 * represented as an {@link org.apache.calcite.util.ImmutableBitSet}, where each bit position
 * represents a 0-based output column ordinal.
 *
 * <p>Different from the unique keys: In distributed streaming computing, one record may be
 * divided into RowKind.UPDATE_BEFORE and RowKind.UPDATE_AFTER. If a key changing join is
 * connected downstream, the two records will be divided into different tasks, resulting in
 * disorder. In this case, the downstream cannot rely on the order of the original key. So in
 * this case, it has unique keys in the traditional sense, but it doesn't have change log upsert
 * keys.
 *
 * @return set of keys, or null if this information cannot be determined (whereas empty set
 *     indicates definitely no keys at all)
 */
public Set<ImmutableBitSet> getChangeLogUpsertKeys(RelNode rel);