-- 1. Pause table data refresh
ALTER MATERIALIZED TABLE dwd_orders SUSPEND;

-- 2. Resume table data refresh
ALTER MATERIALIZED TABLE dwd_orders RESUME
-- Set table option via WITH clause
WITH(
  'sink.parallesim' = '10'
);

-- Refresh historical partition
ALTER MATERIALIZED TABLE dwd_orders REFRESH PARTITION(ds='20231023')

UPDATE dwd_orders SET order_number = 10 WHERE id = 73635185;

ALTER MATERIALIZED TABLE dwd_orders SET FRESHNESS = INTERVAL '1' DAY

CREATE MATERIALIZED TABLE dws_deals(
	PRIMARY KEY (order_id, ds) NOT ENFORCED
)
PARTITIONED BY (ds)
FRESHNESS = INTERVAL '1' HOUR
AS SELECT
  o.ds
  o.order_id,
  count(*) order_cnt,
  sum(pay_amt) order_total_amt
  ...
FROM 
  dwd_orders as o
group by o.order_id, o.ds

SELECT * FROM dwd_orders WHERE ds = '20231023' LIMIT 10;

CREATE MATERIALIZED TABLE [catalog_name.][db_name.]table_name

[ ([ <table_constraint> ]) ]

[COMMENT table_comment]

[PARTITIONED BY (partition_column_name1, partition_column_name2, ...)]

[WITH (key1=val1, key2=val2, ...)]

FRESHNESS = INTERVAL '<num>' { SECOND | MINUTE | HOUR | DAY }

[REFRESH_MODE = { CONTINUOUS | FULL }]

AS <select_statement>

<table_constraint>:
  [CONSTRAINT constraint_name] PRIMARY KEY (column_name, ...) NOT ENFORCED

ALTER MATERIALIZED TABLE [catalog_name.][db_name.]table_name SUSPEND

ALTER MATERIALIZED TABLE [catalog_name.][db_name.]table_name RESUME

[WITH('key1' = 'val1', 'key2' = 'val2')]

ALTER MATERIALIZED TABLE [catalog_name.][db_name.]table_name REFRESH

[PARTITION (key1=val1, key2=val2, ...)]

ALTER MATERIALIZED TABLE [catalog_name.][db_name.]table_name 

  SET FRESHNESS = INTERVAL '<num>' { SECOND | MINUTE | HOUR | DAY }

ALTER MATERIALIZED TABLE  [catalog_name.][db_name.]table_name 

  SET REFRESH_MODE = { FULL | CONTINUOUS }

ALTER MATERIALIZED TABLE [catalog_name.][db_name.]table_name  SET ('key' = 'val');

ALTER MATERIALIZED TABLE [catalog_name.][db_name.]table_name RESET ('key');

DROP MATERIALIZED TABLE [ IF EXISTS ] [catalog_name.][db_name.]table_name

{ DESCRIBE | DESC } JOB ${jobId} 

/**
 * Represents the unresolved metadata of a materialized table in a {@link Catalog}.
 *
 * <p>Materialized Table definition: In the context of integrated stream-batch data storage, it provides
 * full history data and incremental changelog. By defining the data's production business logic and
 * freshness, data update is achieved through continuous or full refresh mode, while also possessing
 * the capability for both batch and incremental consumption.
 *
 * <p>The metadata for {@link CatalogMaterializedTable} also includes the following four main parts:
 *
 * <ul>
 *   <li>Schema, comments, options and partition keys.
 *   <li>Data freshness, which determines when the data is generated and becomes visible for user.
 *   <li>Data production business logic, also known as the definition query.
 *   <li>Background refresh pipeline, either through a flink streaming or scheduled batch job, it is
 *       initialized after materialized table is created.
 * </ul>
 * <p>A catalog implementer can either use {@link #of(Schema, String, List, Map, Long, String,
 * Duration, LogicalRefreshMode, RefreshMode, RefreshStatus, String, byte[])} for a basic
 * implementation of this interface or create a custom class that allows passing catalog-specific
 * objects all the way down to the connector creation (if necessary).
*/
@PublicEvolving
public interface CatalogMaterializedTable extends CatalogBaseTable {

    @Override
    default TableKind getTableKind() {
        return TableKind.MATERIALIZED_TABLE;
    }

    /**
     * Check if the table is partitioned or not.
     *
     * @return true if the table is partitioned; otherwise, false
     */
    boolean isPartitioned();

    /**
     * Get the partition keys of the table. This will be an empty set if the table is not
     * partitioned.
     *
     * @return partition keys of the table
     */
    List<String> getPartitionKeys();

    /**
     * Returns a copy of this {@code CatalogMaterializedTable} with given table options {@code options}.
     *
     * @return a new copy of this table with replaced table options
     */
    CatalogMaterializedTable copy(Map<String, String> options);

    /**
     * Returns a copy of this {@code CatalogDynamicTable} with given refresh info.
     *
     * @return a new copy of this table with replaced refresh info
     */
    CatalogDynamicTable copy(
            RefreshStatus refreshStatus,
            String refreshHandlerDescription,
            byte[] serializedRefreshHandler);
      
    /** Return the snapshot specified for the table. Return Optional.empty() if not specified. */
    Optional<Long> getSnapshot();

    /**
     * The definition query text of materialized table, text is expanded in contrast to the original SQL.
     * This is needed because the context such as current DB is lost after the session, in which
     * view is defined, is gone. Expanded query text takes care of this, as an example.
     *
     * <p>For example, for a dynamic table that is defined in the context of "default" database with
     * a query {@code select * from test1}, the expanded query text might become {@code select
     * `test1`.`name`, `test1`.`value` from `default`.`test1`}, where table test1 resides in
     * database "default" and has two columns ("name" and "value").
     *
     * @return the materialized table definition in expanded text.
     */
    String getDefinitionQuery();

    /** Get the freshness of materialized table which is used to determine the physical refresh mode. */
    Duration getFreshness();

    /** Get the logical refresh mode of materialized table. */
    LogicalRefreshMode getLogicalRefreshMode();

    /** Get the physical refresh mode of materialized table. */
    RefreshMode getRefreshMode();

    /** Get the refresh status of materialized table. */
    RefreshStatus getRefreshStatus();

    /** Return summary description of refresh handler. */
    Optional<String> getRefreshHandlerDescription();

    /**
     * Return the serialized refresh handler of materialized table. This will not be used for describe
     * table.
     */
    byte[] getSerializedRefreshHandler();

    /** The logical refresh mode of materialized table. */
    @PublicEvolving
    enum LogicalRefreshMode {
        /**
         * The refresh pipeline will be executed in continuous mode, corresponding to {@link
         * RefreshMode#CONTINUOUS}.
         */
        CONTINUOUS,

        /**
         * The refresh pipeline will be executed in full mode, corresponding to {@link
         * RefreshMode#FULL}.
         */
        FULL,

        /**
         * The refresh pipeline mode is determined by freshness of materialized table, either {@link
         * RefreshMode#FULL} or {@link RefreshMode#CONTINUOUS}.
         */
        AUTOMATIC
    }

    /** The physical refresh mode of materialized table. */
    @PublicEvolving
    enum RefreshMode {
        CONTINUOUS,
        FULL
    }

    /** Background refresh pipeline status of materialized table. */
    @PublicEvolving
    enum RefreshStatus {
        INITIALIZING,
        ACTIVATED,
        SUSPENDED
    }
}

@PublicEvolving
public interface Catalog {
    
    /**
     * Returns a factory for creating instances from catalog objects.
     *
     * <p>This method enables bypassing the discovery process. Implementers can directly pass
     * internal catalog-specific objects to their own factory. For example, a custom {@link
     * CatalogTable} can be processed by a custom {@link DynamicTableFactory}.
     *
     * <p>If this catalog support to create materialized table, you should also override this method to
     * provide {@link DynamicTableFactory} which help planner to find {@link DynamicTableSource} and
     * {@link DynamicTableSink} correctly during compile optimization phase. If you don't override
     * this method, you must specify the physical connector identifier that this catalog represents
     * storage when create materialized table. Otherwise, the planner can't find the {@link
     * DynamicTableFactory}.
     *
     * <p>Because all factories are interfaces, the returned {@link Factory} instance can implement
     * multiple supported extension points. An {@code instanceof} check is performed by the caller
     * that checks whether a required factory is implemented; otherwise the discovery process is
     * used.
     */
    default Optional<Factory> getFactory() {
        return Optional.empty();
    }
}

/**
 * This interface represents the meta information of current dynamic table background refresh
 * pipeline. The refresh mode maybe continuous or full. The format of the meta information in the
 * two modes is not consistent, so user need to implementation this interface according to .
 *
 * <p>This meta information will be serialized to bytes by {@link RefreshHandlerSerializer}, then
 * store to Catalog for suspend or drop {@link CatalogDynamicTable}.
 *
 * <p>In continuous mode, the format of the meta information maybe { "clusterType": "yarn",
 * "clusterId": "xxx", "jobId": "yyyy" }.
 *
 * <p>In full mode, the meta information format maybe { "endpoint": "xxx", "workflowId": "yyy" }.
 * Due to you may use different workflow scheduler plugin in this mode, you should implement this
 * interface according to your plugin.
 */
@PublicEvolving
public interface RefreshHandler {

    /** Returns a string that summarizes this refresh handler meta information for printing to a console or log. */
    String asSummaryString();
}

/**
 * A validated {@link CatalogMaterializedTable} that is backed by the original metadata coming from the
 * {@link Catalog} but resolved by the framework.
 *
 * <p>Note: This will be converted to {@link ResolvedCatalogTable} by framework during planner
 * optimize query phase.
 */
@PublicEvolving
public class ResolvedCatalogMaterializedTable
        implements ResolvedCatalogBaseTable<CatalogMaterializedTable>, CatalogMaterializedTable {

    private final CatalogMaterializedTable origin;

    private final ResolvedSchema resolvedSchema;

    public ResolvedCatalogMaterializedTable(CatalogMaterializedTable origin, ResolvedSchema resolvedSchema) {
        this.origin =
                Preconditions.checkNotNull(
                        origin, "Original catalog materialized table must not be null.");
        this.resolvedSchema =
                Preconditions.checkNotNull(resolvedSchema, "Resolved schema must not be null.");
    }

    @Override
    public Map<String, String> getOptions() {
        return origin.getOptions();
    }

    @Override
    public String getComment() {
        return origin.getComment();
    }

    @Override
    public CatalogBaseTable copy() {
        return new ResolvedCatalogMaterializedTable((CatalogMaterializedTable) origin.copy(), resolvedSchema);
    }

    @Override
    public CatalogMaterializedTable copy(
            RefreshStatus refreshStatus,
            String refreshHandlerDescription,
            byte[] serializedRefreshHandler) {
        return new ResolvedCatalogMaterializedTable(
                origin.copy(refreshStatus, refreshHandlerDescription, serializedRefreshHandler),
                resolvedSchema);
    }

    @Override
    public Optional<String> getDescription() {
        return origin.getDescription();
    }

    @Override
    public Optional<String> getDetailedDescription() {
        return origin.getDetailedDescription();
    }

    @Override
    public boolean isPartitioned() {
        return origin.isPartitioned();
    }

    @Override
    public List<String> getPartitionKeys() {
        return origin.getPartitionKeys();
    }

    @Override
    public ResolvedCatalogMaterializedTable copy(Map<String, String> options) {
        return new ResolvedCatalogMaterializedTable(origin.copy(options), resolvedSchema);
    }

    @Override
    public Optional<Long> getSnapshot() {
        return Optional.empty();
    }

    @Override
    public CatalogMaterializedTable getOrigin() {
        return origin;
    }

    @Override
    public ResolvedSchema getResolvedSchema() {
        return resolvedSchema;
    }

    @Override
    public String getDefinitionQuery() {
        return origin.getDefinitionQuery();
    }

    @Override
    public Duration getFreshness() {
        return origin.getFreshness();
    }

    @Override
    public LogicalRefreshMode getLogicalRefreshMode() {
        return origin.getLogicalRefreshMode();
    }

    @Override
    public RefreshMode getRefreshMode() {
        return origin.getRefreshMode();
    }

    @Override
    public RefreshStatus getRefreshStatus() {
        return origin.getRefreshStatus();
    }

    @Override
    public Optional<String> getRefreshHandlerDescription() {
        return origin.getRefreshHandlerDescription();
    }

    @Nullable
    @Override
    public byte[] getSerializedRefreshHandler() {
        return origin.getSerializedRefreshHandler();
    }
}

CREATE MATERIALIZED TABLE sink
PARTITIONED BY (ds)
FRESHNESS = INTERVAL '3' MINUTE
AS
-- use cte
WITH cte_source AS (
  SELECT 
  user_id,
  shop_id,
  FROM_UNIXTIME(ds, 'yyyy-MM-dd') AS ds,
  SUM (payment_amount_cents) AS payed_buy_fee_sum,
  SUM (1) AS pv
FROM c_source
GROUP BY user_id, shop_id, FROM_UNIXTIME(ds, 'yyyy-MM-dd')
)
SELECT * FROM cte_source

-- create a materialized table with two agg indices
CREATE MATERIALIZED TABLE dws_deals(
	PRIMARY KEY (order_id, ds) NOT ENFORCED
)
PARTITIONED BY (ds)
FRESHNESS = INTERVAL '1' HOUR
AS SELECT
  o.ds
  o.order_id,
  count(*) order_cnt,
  sum(pay_amt) order_total_amt
FROM 
  dwd_orders as o
GROUP BY o.order_id, o.ds;

-- add two new agg indices statement
ALTER MATERIALIZED TABLE dws_deals
ADD 
SELECT
  o.ds
  o.order_id,
  count(distinct order_id) dedup_order_cnt, -- deduplicated order cnt
  sum(goods_cnt) tatol_goods_cnt -- total goods cnt
FROM 
  dwd_orders as o
GROUP BY o.order_id, o.ds;