1. Motivation

The ALTER MATERIALIZED TABLE... AS <select_statement> command provides a powerful mechanism to evolve a table's definition. However, its current behavior is rigid: it always stops the current job and starts a new one from the beginning of the source data, discarding all previous state.

While a full reprocess is sometimes unavoidable, specifically when the query optimizer generates a completely new physical plan, forcing this behavior for every evolution is inefficient and costly. For many common evolutions, such as adding a nullable column or making compatible logic changes, re-ingesting historical data is unnecessary.

Users require granular control over this process to optimize for cost, recovery speed, and data correctness. This FLIP proposes enhancing CREATE [OR ALTER] and ALTER MATERIALIZED TABLE by introducing the START_MODE clause. This allows users to explicitly define the data processing window, making table evolution configurable and efficient.

2. Proposed Changes

We propose introducing an optional clause to the CREATE [OR ALTER] MATERIALIZED TABLE... AS <select_statement> and the ALTER MATERIALIZED TABLE... AS <select_statement> commands: START_MODE.

2.1. Core Design Principle: Explicit Data Scope

The central principle of this proposal is to give users control over the Data Processing Window (START_MODE).

This clause defines the point in the source stream(s) from which the new job should begin processing data. It answers the question, "What data should the new logic be applied to?" and avoids unnecessary computational costs for historical data that is no longer needed.

2.2. New SQL Syntax

The CREATE [OR ALTER] MATERIALIZED TABLE command will be enhanced as follows:

CREATE [OR ALTER] MATERIALIZED TABLE [catalog_name.][db_name.]table_name
[( { <schema_definition> | <column_list> } )]
[WITH (...)]
[START_MODE = FROM_BEGINNING 
             | FROM_NOW[(<interval_expression>)] 
             | FROM_TIMESTAMP(<timestamp_literal>)
             | RESUME_OR_FROM_BEGINNING 
             | RESUME_OR_FROM_NOW[(<interval_expression>)] 
             | RESUME_OR_FROM_TIMESTAMP(<timestamp_literal>)]
AS <select_statement>

The ALTER MATERIALIZED TABLE command will be enhanced as follows:

-- Explicit alteration of the definition query with reprocessing control
ALTER MATERIALIZED TABLE [catalog_name.][db_name.]table_name
[WITH (...)]
[SET START_MODE = FROM_BEGINNING 
             | FROM_NOW[(<interval_expression>)] 
             | FROM_TIMESTAMP(<timestamp_literal>)
             | RESUME_OR_FROM_BEGINNING 
             | RESUME_OR_FROM_NOW[(<interval_expression>)] 
             | RESUME_OR_FROM_TIMESTAMP(<timestamp_literal>)]
AS <select_statement>

2.3. START_MODE Clause

This clause controls the data processing window for the evolution.

Implementation Note on Time Expressions and RESUME_OR_...: This capability relies on the Catalog and Connector implementation. Timestamp and Interval expressions are resolved by the planner into a standard long timestamp (milliseconds since 1970-01-01 00:00:00 UTC) before being passed to the Catalog/Connector layer, aligning with the existing Catalog.getTable(ObjectPath, long) API for temporal operations.2.4. Defaults and Interaction

To maintain backward compatibility and ensure data correctness by default, the clause has a simple, explicit default.

• START_MODE Default: If START_MODE is not specified, it defaults to FROM_BEGINNING. (Note: The original text had conflicting defaults; setting this to FROM_BEGINNING ensures it exactly mimics the original, fixed behavior of the ALTER command).

This is crucial for backward compatibility. It ensures that any ALTER command run without this new clause will, by default, reprocess the entire history from the source.

2.4. Idempotency and SHOW CREATE [OR ALTER]

To address the non-idempotent nature of relative intervals, the SHOW CREATE [OR ALTER] MATERIALIZED TABLE command will render the user's original expression but include a comment with the timestamp resolved at execution time.

Example SHOW CREATE OR ALTER at submission:

CREATE OR ALTER MATERIALIZED TABLE filtered_high_value_orders
...
START_MODE = FROM_NOW(INTERVAL '7' DAY) /* Evaluated to FROM_TIMESTAMP(TIMESTAMP '2025-10-28 10:00:00') at execution */
AS SELECT...

3. Public Interfaces

To support the persistence of the new evolution clause, the primary public interface to be updated is org.apache.flink.table.catalog.CatalogMaterializedTable.

The new API will rely on a new public, immutable class, StartMode, and its associated enum (StartModeType), which would be added to the org.apache.flink.table.catalog package. The new method will be added as a default method returning Optional.empty() to ensure full backward compatibility with any existing custom catalog implementations.

public interface CatalogMaterializedTable extends CatalogBaseTable {
    //...
  
    /**
     * Returns the structured object for the {@code START_MODE} clause.
     *
     * <p>This defines the point in the source stream(s) from which a new job should begin
     * processing data upon creation or evolution.
     *
     * <p>If {@link Optional#empty()}, the system default ({@code FROM_BEGINNING}) will be
     * used by the planner.
     *
     * @return The structured StartMode object.
     */
    default Optional<StartMode> getStartMode() {
      return Optional.empty();
    }

    //...
}

4. User Journeys

The following table outlines the most common user journeys and how the START_MODE clause achieves them. (Note: In all current scenarios, existing state is discarded, matching Flink's default behavior).

5. Compatibility, Deprecation, and Migration Plan

This proposal is fully backward compatible. The START_MODE clause is optional.

If omitted, it defaults to START_MODE = FROM_BEGINNING. This exactly matches the existing, unmodified behavior of the ALTER command. No migration is needed for existing materialized tables or user scripts. Existing ALTER scripts will continue to function as they always have.

6. Rejected Alternatives

• Single REPROCESS Clause: This was rejected because it did not provide a clear enough taxonomy for the different ways a user might want to resume or start data consumption. START_MODE is more explicit and aligns better with stream processing concepts.

7. Test Plan

The implementation will be validated with unit and integration tests covering:

• Parser: Ensure the queries are parsed correctly both with and without the new clause.

• Planner/Execution: Verify that a materialized table created or altered follows the described behavior above.

• End-to-End Tests: Add tests to validate the user journeys described above.