DELETE FROM user WHERE id = -1;
DELETE FROM user WHERE id > (SELECT count(*) FROM employee);

UPDATE user SET name = "u1" WHERE id > 10;
UPDATE user SET name = "u1" WHERE id > (SELECT count(*) FROM employee);

/**
 * The context for scan the table to do row-level modification like updated/delete, designed to provide some necessary
 * information that the sink may need to know to perform update/delete. It'll be generated by table
 * source which implements {@link SupportsRowLevelModificationScan}, and then passed to the sink
 * which implements {@link SupportsRowLevelDelete} or {@link SupportsRowLevelUpdate} for
 * UPDATE/DELETE statement in compile phase.
 *
 * <p>This mechanism enables the coordination between the source and sink for the corresponding
 * table to be updated/deleted.
 *
 * <p>The connector can implement this interface to provide custom information.
 */
@PublicEvolving
public interface RowLevelModificationScanContext {}

/**
 * Enables to tell the {@link ScanTableSource} the type of row-level modification for the table, and
 * return {@link RowLevelModificationScanContext} which will then be passed to the sink which
 * implements {@link SupportsRowLevelDelete} or {@link SupportsRowLevelUpdate}. It allows the table
 * source to pass some values to the table sink which is to be updated/deleted.
 *
 * <p>Note: This interface is optional for table source to support update/delete existing data. For
 * the case that the table source won't need to know the type of row-level modification or pass some
 * values to sink, the table source doesn't need to implement it.
 *
 * <p>Note: Only if the {@link ScanTableSource} implements this interface, the method {@link #applyRowLevelModificationScan} will
 * be involved. For more details, please see the method {@link #applyRowLevelModificationScan}.
 */
public interface SupportsRowLevelModificationScan {

    /**
     * Apply the type of row-level modification and the previous {@link
     * RowLevelModificationScanContext} returned by previous table source scan, return a new {@link
     * RowLevelModificationScanContext} which will then finally be passed to the table sink.
     *
     * <p>Note: for the all tables in the UPDATE/DELETE statement, this method will be involved for
     * the corresponding table source scan.
     *
     * <p>Note: it may have multiple table sources in the case of sub-query. In such case, it will
     * return multiple {@link RowLevelModificationScanContext}. To handle such case, the planner
     * will also pass the previous {@link RowLevelModificationScanContext} to the current table
     * source scan to make it decide how to deal with the previous {@link
     * RowLevelModificationScanContext}. The order is consistent to the order that the table source
     * compiles. The planner will only pass the last context returned to the sink.
     *
     * @param previousContext the context returned by previous table source, if there's no previous
     *     context, it'll be null.
     */
    RowLevelModificationScanContext applyRowLevelModificationScan(
            RowLevelModificationType rowLevelModificationType,
            @Nullable RowLevelModificationScanContext previousContext);

    /**
     * Type of the row-level modification for the table.
     *
     * <p>Currently, two types are supported:
     *
     * <ul>
     *   <li>UPDATE
     *   <li>DELETE
     * </ul>
     */
    enum RowLevelModificationType {
        UPDATE,
        DELETE
    }
}

/**
 * Enable to delete existing data in a {@link DynamicTableSink} directly according to the filter
 * expressions in {@code DELETE} clause.
 *
 * <p>Given the following SQL:
 *
 * <pre>{@code
 * DELETE FROM t WHERE (a = '1' OR a = '2') AND b IS NOT NULL;
 *
 * }</pre>
 *
 * <p>In the example above, {@code [a = '1' OR a = '2']} and {@code [b IS NOT NULL]} are the
 * filters.
 *
 * <p>Flink will get the filters in conjunctive form and push down the filters into sink by calling
 * method {@link #applyDeleteFilters} in the planner phase. If it returns true, Flink will then call
 * method {@link #executeDeletion} to execute the actual delete in execution phase.
 *
 * <p>Note: in the case that the filter expression is not available, e.g., sub-query or {@link
 * #applyDeleteFilters} returns false, if the sink implements {@link SupportsRowLevelDelete}, Flink
 * will try to rewrite the delete operation and produce row-level changes, see {@link
 * SupportsRowLevelDelete} for more details. Otherwise, Flink will throw {@link
 * UnsupportedOperationException}.
 */
@PublicEvolving
public interface SupportsDeletePushDown {

    /**
     * Provides a list of filters from delete operation in conjunctive form in planning phase. A
     * sink can either return true if it can accept all filters or return false if it can not
     * accept.
     *
     * <p>If it returns true, Flink will then call the method {@link #executeDeletion} in execution
     * phase to do the actual deletion.
     *
     * <p>If it returns false, and the sink still implements {@link SupportsRowLevelDelete}, Flink
     * will rewrite the delete operation and produce row-level changes. Otherwise, Flink will throw
     * {@link UnsupportedOperationException}.
     */
    boolean applyDeleteFilters(List<ResolvedExpression> filters);

    /**
     * Do the actual deletion in the execution phase, and return how many rows has been deleted,
     * Optional.empty() is for unknown delete rows.
     *
     * <p>Note that this method will be involved if and only if {@link #applyDeleteFilters(List
     * ResolvedExpression)} returns true. So, please make sure the implementation for this method
     * will do delete the data correctly.
     *
     * <p>Note that in this method, the sink won't get the filters since they have been passed to
     * the method {@link #applyDeleteFilters} before. The sink may need to keep these filters, so
     * that it can get the filters if necessary to finish the deletion.
     */
    Optional<Long> executeDeletion();
}

/**
 * Enable to delete existing data in a {@link DynamicTableSink} according to row-level changes.
 *
 * <p>The planner will call method {@link #applyRowLevelDelete} to get the {@link
 * RowLevelDeleteInfo} that the sink returns, and rewrite the delete operation based on gotten
 * {@link RowLevelDeleteInfo} to produce rows (may rows to be deleted or remaining rows after the
 * delete operation depending on sink's implementation) to {@link DynamicTableSink}. The sink is
 * expected to consume these rows to achieve rows delete purpose.
 *
 * <p>Note: if the sink also implements {@link SupportsDeletePushDown}, the planner will always
 * prefer {@link SupportsDeletePushDown}, and only the filters aren't available or {@link
 * SupportsDeletePushDown#applyDeleteFilters} returns false, this interface will be considered and
 * to rewrite the delete operation to produce the rows to the sink.
 */
@PublicEvolving
public interface SupportsRowLevelDelete {


     /**
     * Apply row level delete with {@link RowLevelModificationScanContext} passed by table source, and
     * return {@link RowLevelDeleteInfo} to guide the planner on how to rewrite the delete
     * operation.
     *
     * <p>Note: if the table source doesn't implement {@link SupportsRowLevelModificationScan}, the
     * {@param context} will be null.
     */
    RowLevelDeleteInfo applyRowLevelDelete(@Nullable RowLevelModificationScanContext context);
   

    /** The information that guides the planer on how to rewrite the delete operation. */
    @PublicEvolving
    interface RowLevelDeleteInfo {

        /**
         * The required columns that the sink expects for deletion, the rows consumed by sink will
         * contain the columns with the order consistent with the order of returned columns. If
         * return Optional.empty(), it will select all columns.
         */
        default Optional<List<Column>> requiredColumns() {
            return Optional.empty();
        }

        /**
         * Planner will rewrite delete to query base on the {@link RowLevelDeleteInfo}, keeps the
         * query of delete unchanged by default(in `DELETE_ROWS` mode), or change the query to the
         * complementary set in REMAINING_ROWS mode.
         *
         * <p>Take the following SQL as an example:
         *
         * <pre>{@code
         * DELETE FROM t WHERE y = 2;
         * }</pre>
         *
         * <p>If returns {@link RowLevelDeleteMode#DELETED_ROWS}, the sink will get rows to be
         * deleted which match the filter [y = 2].
         *
         * <p>If returns {@link RowLevelDeleteMode#REMAINING_ROWS}, the sink will get the rows which
         * doesn't match the filter [y = 2].
         *
         * <p>Note: All rows will have RowKind#DELETE when RowLevelDeleteMode is DELETED_ROWS, and
         * RowKind#INSERT when RowLevelDeleteMode is REMAINING_ROWS.
         */
        default RowLevelDeleteMode getRowLevelDeleteMode() {
            return RowLevelDeleteMode.DELETED_ROWS;
        }
    }
    /**
     * Type of delete modes that the sink expects for delete purpose.
     *
     * <p>Currently, two modes are supported:
     *
     * <ul>
     *   <li>DELETED_ROWS - in this mode, the sink will only get the rows that need to be
     *       deleted.
     *   <li>REMAINING_ROWS - in this mode, the sink will only get the remaining rows as if the
     *       the delete operation had been done.
     * </ul>
     */
    @PublicEvolving
    enum RowLevelDeleteMode {
            DELETED_ROWS,
            REMAINING_ROWS
    }
}

/**
 * Enable to update existing data in a {@link DynamicTableSink} according to row-level changes.
 *
 * <p>The planner will call method {@link #applyRowLevelUpdate} to get the {@link
 * RowLevelUpdateInfo} that the sink returns, and rewrite the update operation based on gotten
 * {@link RowLevelUpdateInfo} to produce rows (may rows to be updated with update after values or
 * all rows including both rows be updated and the other rows without update need depending on
 * sink's implementation) to {@link DynamicTableSink}. The sink is expected to consume these rows to
 * achieve rows update purpose.
 */
@PublicEvolving
public interface SupportsRowLevelUpdate {

    /**
     * Apply row level update with providing the updated columns and {@link
     * RowLevelModificationScanContext} passed by table source, then return {@link RowLevelUpdateInfo}
     * to guide the planner on how to rewrite the update operation.
     *
     * @param updatedColumns the columns updated by update operation in table column order.
     * @param rowLevelModificationScanContext the context passed by table source which implement {@link
     *     SupportsRowLevelModificationScan}, if the table source doesn't implement this interface,
     *     the rowLevelModificationContext will be null.
     */
    RowLevelUpdateInfo applyRowLevelUpdate(List<Column> updatedColumns, @Nullable RowLevelModificationScanContext context);

    /** The information that guides the planer on how to rewrite the update operation. */
    @PublicEvolving
    interface RowLevelUpdateInfo {

        /**
         * The required columns that the sink expects for deletion, the rows consumed by sink will
         * contain the columns with the order consistent with the order of returned columns. If
         * return Optional.empty(), it will select all columns.
         */
        default Optional<List<Column>> requiredColumns() {
            return Optional.empty();
        }

        /**
         * Planner will rewrite the update operation to query base on the {@link
         * RowLevelUpdateMode}, keeps the query of update unchanged by default(in `UPDATED_ROWS`
         * mode), or change the query to union the updated rows and un-updated rows (in `ALL_ROWS`
         * mode).
         *
         * <p>Take the following SQL as an example:
         *
         * <pre>{@code
         * UPDATE t SET x = 1 WHERE y = 2;
         * }</pre>
         *
         * <p>If returns {@link RowLevelUpdateMode#UPDATED_ROWS}, the sink will get the update after
         * rows which match the filter [y = 2].
         *
         * <p>If returns {@link RowLevelUpdateMode#ALL_ROWS}, the sink will get the update after
         * rows which match the filter [y = 2] as well as the other rows that don't match the filter
         * [y = 2].
         *
         * <p>Note: All rows will have RowKind#UPDATE_AFTER when RowLevelUpdateMode is UPDATED_ROWS,
         * and RowKind#INSERT when RowLevelUpdateMode is ALL_ROWS.
         */
        default RowLevelUpdateMode getRowLevelUpdateMode() {
            return RowLevelUpdateMode.UPDATED_ROWS;
        }
    }
    /**
     * Type of update modes that the sink expects for update purpose.
     *
     * <p>Currently, two modes are supported:
     *
     * <ul>
     *   <li>UPDATED_ROWS - in this mode, The sink will only get the rows that need to updated
     *       with updated value.
     *   <li>ALL_ROWS - in this mode, the sink will get all the rows containing the updated rows
     *       and the other rows that don't need to be updated.
     * </ul>
     */
    @PublicEvolving
    enum RowLevelUpdateMode {
        UPDATED_ROWS,
        ALL_ROWS
    }
}

DELETE FROM user WHERE id = -1;

SELECT id, name FROM user WHERE id = -1;

SELECT id FROM user WHERE id = -1;

DELETE FROM user WHERE id = -1;

SELECT id, name
FROM user
WHERE NOT(id = -1);

UPDATE user SET name = "u1" WHERE id > 10;

SELECT id, "u1" FROM user WHERE id > 10;

UPDATE user SET name = "u1" WHERE id > 10;

SELECT id, IF(id > 10, "u1", name) FROM user;

public class TSink implements DynamicTableSink, SupportsRowLevelDelete {
  
  ....
  
  @Override
  public SinkRuntimeProvider getSinkRuntimeProvider(Context context) {
  return new DataStreamSinkProvider() {
    @Override
    public DataStreamSink<?> consumeDataStream(
        ProviderContext providerContext, DataStream<RowData> dataStream) {
          // consume the rows for delete purpose
        }
    };
   }
}

public class TSink implements DynamicTableSink, SupportsDelete {
  
  ....
  
  @Override
  public SinkRuntimeProvider getSinkRuntimeProvider(Context context) {
  return new DataStreamSinkProvider() {
    @Override
    public DataStreamSink<?> consumeDataStream(
        ProviderContext providerContext, DataStream<RowData> dataStream) {
          // consume the rows for delete purpose
        }
    };
   }
}

-- Custom a ModificationScanContext
public MyTableModificationScanContext implements RowLevelModificationScanContext {
    privite Map<ObjectIdentifer, Long> snapshots = new HashMap<>();
     
    Map<ObjectIdentifer, Long> getSnapshots() {
      return snapshots;
    }
}

-- Source
public class TSource
        implements ScanTableSource, SupportsRowLevelModificationScan {
    privite RowLevelModificationType type;

    RowLevelModificationScanContext applyRowLevelModificationScan(
            RowLevelModificationType rowLevelModificationType,
            @Nullable RowLevelModificationScanContext previousContext) {

    if (previousContext == null) {
     // no previous context, first table source 
     previousContext = new MyTableModificationScanContext();
    }
 
    // mark the scan type
     this.type = rowLevelModificationType;
     // put some values
     scanContext.getSnapshots().put(xxx, xxx);
   }
 }

-- Sink 
public class TSink
        implements DynamicTableSink, SupportsRowLevelDelete {
  
     RowLevelDeleteInfo applyRowLevelDelete(@Nullable RowLevelModificationScanContext context) {
       MyTableModificationScanContext myContext = (MyTableModificationScanContext) context;
       // get the snapshot info
       myContext.getSnapshots().get(xxx);
     }
}