-- Source Input
source table1：id1（primary key），f1，f2，f3，null，null，null，null，null，null，... ...
source table2：id1（primary key），null，null，null，f4，f5，f6，null，null，null，... ...
source table3：id1（primary key），null，null，null，null，null，null，f7，f8，f9，... ... 
... ...
source table5：id1（primary key），... ...


-- Sink Output
sink table：id1，f1，f2，f3，f4，f5，f6，f7，f8，f9，... ...


-- Flink SQL  
INSERT INTO sink
SELECT  id,
        f1,
        f2,
        f3,
        f4,
        f5,
        f6,
        f7,
        f8,
        f9,
  ... ...
FROM (
    SELECT  id,
            f1,
            f2,
            f3,
            CAST(NULL as STRING) as f4,
            CAST(NULL as STRING) as f5,
            CAST(NULL as STRING) as f6,
            CAST(NULL as STRING) as f7,
            CAST(NULL as STRING) as f8,
            CAST(NULL as STRING) as f9,
      		... ...
    FROM  table1
    UNION ALL
    SELECT  id,
            CAST(NULL as STRING) as f1,
            CAST(NULL as STRING) as f2,
            CAST(NULL as STRING) as f3,
            f4,
            f5,
            f6,
            CAST(NULL as STRING) as f7,
            CAST(NULL as STRING) as f8,
            CAST(NULL as STRING) as f9,
      		... ...
    FROM  table2
   	UNION ALL
    SELECT  id,
            CAST(NULL as STRING) as f1,
            CAST(NULL as STRING) as f2,
            CAST(NULL as STRING) as f3,
            CAST(NULL as STRING) as f4,
            CAST(NULL as STRING) as f5,
            CAST(NULL as STRING) as f6,
            f7,
            f8,
            f9,
  	 		... ...
    FROM table3
    UNION ALL
   	SELECT
    		... ...
  	FROM table4 
    UNION ALL
   	SELECT
    		... ...
  	FROM table5 
)

BEGIN STATEMENT SET;

INSERT INTO sink(pk, f1, f2, f3) 
SELECT 
	pk, 
	f1, 
	f2, 
	f3 
FROM table1;

INSERT INTO sink(pk, f4, f5, f6) 
SELECT 
	pk, 
	f4, 
	f5, 
	f6 
FROM table2;

INSERT INTO sink(pk, f7, f8, f9) 
SELECT 
	pk, 
	f7, 
	f8, 
	f9 
FROM table3;

... ...


INSERT INTO sink(pk, ... ...) 
SELECT 
	pk, 
	... ...
FROM table5;

END;

    @PublicEvolving
    interface Context {

		... ...

        /**
         *
         * @deprecated use {@link
         *     org.apache.flink.table.connector.sink.abilities.SupportsTargetColumnWriting} instead.
         */         
		@Deprecated(since = "2.1")
        Optional<int[][]> getTargetColumns();
    }

package org.apache.flink.table.connector.sink.abilities;

/**
 * Interface for {@link DynamicTableSink}s that support target column writing.
 *
 * <p>The planner will parse target columns from the DML clause and call {@link
 * #applyTargetColumns(int[][])} to pass an array of column index paths to sink.
 *
 * <p>The array indices are 0-based and support composite columns within (possibly nested)
 * structures. This information comes from the column list of the DML clause, e.g., for a sink table
 * t1 which schema is: {@code a STRING, b ROW < b1 INT, b2 STRING>, c BIGINT}
 *
 * <ul>
 *   <li>insert: 'insert into t1(a, b.b2) ...', the column list will be 'a, b.b2', and will provide
 *       {@code [[0], [1, 1]]}. The statement 'insert into t1 select ...' will not apply this
 *       ability.
 *   <li>update: 'update t1 set a=1, b.b1=2 where ...', the column list will be 'a, b.b1', and will
 *       provide {@code [[0], [1, 0]]}.
 * </ul>
 *
 * <p>Note: Planner will not apply this ability for the delete statement because it has no column
 * list.
 *
 * <p>A sink can use this information to perform target columns writing.
 *
 * <p>If this interface is implemented and {@link #applyTargetColumns(int[][])} returns true. The
 * planner will use this information for plan optimization such as sink reuse.
 */
@PublicEvolving
public interface SupportsTargetColumnWriting {

    /**
     * Provides an array of column index paths related to user specified target column list.
     *
     * <p>See the documentation of {@link SupportsTargetColumnWriting} for more information.
     *
     * @param targetColumns column index paths
     * @return true if the target columns are applied successfully, false otherwise.
     */
    boolean applyTargetColumns(int[][] targetColumns);
} 

package org.apache.flink.table.planner.plan.abilities.sink;


/**
 * A sub-class of {@link SinkAbilitySpec} that can not only serialize/deserialize the writing target
 * column indices to/from JSON, but also can write the target columns for {@link
 * SupportsTargetColumnWriting}.
 */
@JsonIgnoreProperties(ignoreUnknown = true)
@JsonTypeName("TargetColumnWriting")
public class TargetColumnWritingSpec implements SinkAbilitySpec {
    public static final String FIELD_NAME_TARGET_COLUMNS = "targetColumns";

    @JsonProperty(FIELD_NAME_TARGET_COLUMNS)
    private final int[][] targetColumns;

    @JsonCreator
    public TargetColumnWritingSpec(@JsonProperty(FIELD_NAME_TARGET_COLUMNS) int[][] targetColumns) {
        this.targetColumns = targetColumns;
    }

    @Override
    public void apply(DynamicTableSink tableSink) {
        if (tableSink instanceof SupportsTargetColumnWriting) {
            ((SupportsTargetColumnWriting) tableSink).applyTargetColumns(targetColumns);
        } else {
            throw new TableException(
                    String.format(
                            "%s does not support SupportsTargetColumnWriting.",
                            tableSink.getClass().getName()));
        }
    }

    @Override
    public boolean equals(Object o) {
        if (this == o) {
            return true;
        }
        if (o == null || getClass() != o.getClass()) {
            return false;
        }
        TargetColumnWritingSpec that = (TargetColumnWritingSpec) o;
        return Objects.deepEquals(targetColumns, that.targetColumns);
    }

    @Override
    public int hashCode() {
        return Arrays.deepHashCode(targetColumns);
    }
}

public class ReusableSinkGroup {

    private final List<Sink> originalSinks = new ArrayList<>();
    private final SinkAbilitySpec[] sinkAbilitySpecs;
    private final RelTraitSet inputTraitSet;
    private final String digest;
       
		ReusableSinkGroup(
            Sink sink,
            String digest,
            SinkAbilitySpec[] sinkAbilitySpecs,
            RelTraitSet inputTraitSet) {
        this.originalSinks.add(sink);
        this.sinkAbilitySpecs = sinkAbilitySpecs;
        this.inputTraitSet = inputTraitSet;
        this.digest = digest;
    }
}

@PublicEvolving
public interface SupportsSinkReuse {

	
	public boolean equals(Object sink);
}