...
Hence, starting from the concept of unified streaming and batch processing (One SQL & Unified Table), it's time to turn Dynamic Table from an abstract concept into a tangible entity. We propose the Dynamic Table entity to we propose a new table type based on the Dynamic Tables concept[1], called Materialized Table, to simplify streaming and batch ETL pipeline cost-effectively. This allows users to no longer be confused by various complex concepts such as streaming and batch processing, as well as underlying technical details, and to focus solely on business logic and data freshness.
...
Let's walk through a user story to see how Dynamic Materialized Table solves the problem described earlier.
...
Code Block | ||
---|---|---|
| ||
CREATE DYNAMICMATERIALIZED TABLE dwd_orders ( PRIMARY KEY(ds, id) NOT ENFORCED ) PARTITIONED BY (ds) FRESHNESS = INTERVAL '3' MINUTE AS SELECT o.ds o.id, o.order_number, o.user_id, ... FROM orders as o LEFT JOIN products FOR SYSTEM_TIME AS OF proctime() AS prod ON o.product_id = prod.id LEFT JOIN order_pay AS pay ON o.id = pay.order_id and o.ds = pay.ds |
When a Create Dynamic Materialized Table statement is executed, it deduces the Schema schema of the table based on the query statement referenced, first creates a Dynamic Materialized Table in Catalog, and then automatically creates a data processing pipeline to refresh the table based on the specified Freshness value. Since FRESHNESS is set to 3 minutes here, a Flink Streaming Job is pulled up to refresh continuously.
...
Code Block | ||
---|---|---|
| ||
-- 1. Pause table data refresh ALTER DYNAMICMATERIALIZED TABLE dwd_orders SUSPEND; -- 2. Resume table data refresh ALTER DYNAMICMATERIALIZED TABLE dwd_orders RESUME -- Set table option via WITH clause WITH( 'sink.parallesim' = '10' ); |
...
Code Block | ||
---|---|---|
| ||
-- Refresh historical partition ALTER DYNAMICMATERIALIZED TABLE dwd_orders REFRESH PARTITION(ds='20231023') |
...
Code Block | ||
---|---|---|
| ||
ALTER DYNAMICMATERIALIZED TABLE dwd_orders SET FRESHNESS = INTERVAL '1' DAY |
If you want to adjust the data freshness, you just need to modify this dynamic materialized table freshness attribute. It will stop the Flink streaming job first, then create a scheduled task automatically in the workflow scheduler. The workflow will trigger the table refresh by the specified freshness.
From the above operational process, it can be seen that the use of Dynamic Materialized Table can greatly simplify the development of the data processing pipeline. By using a unified declarative SQL API to unify the stream and batch pipelines, users do not need to be concerned with Flink job and execution mode, they simply need to operate with the dynamic materialized table.
In summary, we hope that through the Dynamic Materialized Table with declared SQL statement and freshness, you can define the batch and streaming transformation on your data in the same way, accelerate ETL pipeline development, automatically manage task orchestration, enhance data quality and freshness, save your time and effort.
...
Code Block | ||
---|---|---|
| ||
CREATE DYNAMICMATERIALIZED 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 |
We can create cascading DWS Dynamic Materialized Table dws_deals based on DWD's Dynamic Materialized Table dwd_orders.
OLAP Query
Code Block | ||
---|---|---|
| ||
SELECT * FROM dwd_orders WHERE ds = '20231023' LIMIT 10; |
We can also select the Dynamic Materialized Table from dwd_orders by a point query.
Public Interfaces
SQL Statements
To support dynamic tablesmaterialized table, we need to introduce some new syntax.
...
Code Block | ||
---|---|---|
| ||
CREATE DYNAMICMATERIALIZED 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 |
...
- PRIMARY KEY: An optional list of columns uniquely identifies each row within the table. The column as primary key must be non-null.
- PARTITIONED BY: An optional list of columns of the table to partition the table by.
- WITH: Optionally sets one or more table properties.
- FRESHNESS: Define the time measure that the dynamic materialized table’s content should lag behind updates to the base tables. This is merely a best-effort behavior, not a guarantee.
- REFRESH_MODE:An Optional clause specifies the refresh type for the dynamic materialized table. This property can not be altered after the dynamic materialized table is created.
- AS select_statement: This clause populates the dynamic materialized table using the data from the select query. The base tables can be a dynamic materialized table, table or view.
Limitations:
- Does not support explicitly specifying columns.
- Does not support modifying the select statement after the dynamic materialized table is created.
- Does not support referencing temporary tables, views, or functions in Select statements currently.
FRESHNESS Clause
For a dynamic materialized table, a very important attribute is freshness, so we must highlight its function first. Freshness serves two key functions.
Firstly, freshness is used for determining the data refresh mode of a dynamic materialized table which can satisfy the specified data freshness, there are two types of refresh modes currently:
- Continuous Refresh Mode: Launches a Flink streaming job that continuously refreshes the dynamic materialized table data.
- Full Refresh Mode: A Workflow Scheduler periodically triggers a Flink batch job to refresh the dynamic materialized table data.
Regarding how to determine the refresh mode, you can see the Configuration section for more detail.
...
- In continuous mode, it will be used to derive checkpoint & minibatch related options to determine the visibility of data to satisfy the defined freshness.
- In full mode, it will be used to derive the scheduling interval of the workflow, i.e., the Cron expression, and calculate the corresponding time partition value (e.g., ds) for partitioned table, to satisfy the defined data freshness of dynamic materialized tabe.
REFRESH_MODE Clause
In the previous section we explained that Refresh Mode is automatically derived by the engine based on freshness, so why do we provide the REFRESH_MODE clause to support specifying Refresh Mode manually? What is the relationship between these two? Considering different business scenarios, in some scenarios the user knows exactly what refresh mode is needed, and the refresh mode derived from freshness can't meet the demand, so we provide a more flexible way to allow manual specification of refresh mode.
...
In full refresh mode, for dynamic materialized table, the default overwrite behavior is table granularity. If a dynamic materialized table is a partitioned table, and the time partition field format is specified by 'partition.fields.#.date-formatter' when creating the table, the default overwrite behavior is partition granularity, that is, only the latest partition data is overwritten each time. For more details on 'partition.fields.#.date-formatter', see the partitioned table full refresh section.
Operations on dynamic materialized tables are executed through the following Alter statements.
...
Code Block | ||
---|---|---|
| ||
ALTER DYNAMICMATERIALIZED TABLE [catalog_name.][db_name.]table_name SUSPEND |
Stop or suspend refreshes on the dynamic materialized table, also means that the background refresh job is suspended. If the dynamic materialized table is used by other downstream dynamic tables, the downstream will not be suspended, effective only for the current dynamic materialized table.
Resume
Code Block | ||
---|---|---|
| ||
ALTER DYNAMICMATERIALIZED TABLE [catalog_name.][db_name.]table_name RESUME [WITH('key1' = 'val1', 'key2' = 'val2')] |
...
Restart or resume refresh on the dynamic materialized table, also means that the background refresh job is resumed. Resume operations only work for current dynamic materialized table, and don't cascade resumes downstream dynamic materialized tables are suspended.
You can also specify one or more dynamic materialized table options by WITH keyword to optimize the background refresh job, these options are only effective for the current operation and will not be persisted.
...
Code Block | ||
---|---|---|
| ||
ALTER DYNAMICMATERIALIZED TABLE [catalog_name.][db_name.]table_name REFRESH [PARTITION (key1=val1, key2=val2, ...)] |
Manually trigger a dynamic materialized table refresh, only works for the current table, doesn't trigger a cascade refresh of all downstream tables defaultly.
...
Code Block | ||
---|---|---|
| ||
ALTER DYNAMICMATERIALIZED TABLE [catalog_name.][db_name.]table_name SET FRESHNESS = INTERVAL '<num>' { SECOND | MINUTE | HOUR | DAY } |
Modify the time measure that the dynamic materialized table’s content should lag behind updates to the base tables. Modifying freshness may affect the refresh mode of the background job, which may change from continuous mode to full mode, or from full mode to continuous mode, or remain unchanged. In any case, alter freshness will stop the current refresh job and create a new job with the updated freshness.
Note: If the refresh mode is manually specified when creating a dynamic materialized table, modify freshness will not modify the refresh mode.
...
Code Block | ||
---|---|---|
| ||
ALTER DYNAMICMATERIALIZED TABLE [catalog_name.][db_name.]table_name SET REFRESH_MODE = { FULL | CONTINUOUS } |
Modify the refresh mode of the dynamic materialized table, which may change from continuous mode to full mode, or full mode to continuous mode.
...
Code Block | ||
---|---|---|
| ||
ALTER DYNAMICMATERIALIZED TABLE [catalog_name.][db_name.]table_name SET ('key' = 'val'); |
...
Code Block | ||
---|---|---|
| ||
ALTER DYNAMICMATERIALIZED TABLE [catalog_name.][db_name.]table_name RESET ('key'); |
Besides the aforementioned Alter syntax, dynamic materialized tables currently do not support other Alter syntaxes that are applicable to regular Table.
...
Code Block | ||
---|---|---|
| ||
DROP DYNAMICMATERIALIZED TABLE [ IF EXISTS ] [catalog_name.][db_name.]table_name |
Drop dynamic materialized table will delete the refresh job first, then drop the metadata and data.
...
We introduce a new interface CatalogDynamicTableCatalogMaterializedTable which extends the CatalogBaseTable. As described above, the CatalogDynamicTableCatalogMaterializedTable, in addition to having the attributes of schema, options and partition keys , also encompasses three other important metadata: definition query, freshness, refresh mode, refresh status and refresh jobhandler.
Code Block | ||
---|---|---|
| ||
/** * Represents the unresolved metadata of a dynamicmaterialized table in a {@link Catalog}. * * <p>Dynamic<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 CatalogDynamicTableCatalogMaterializedTable} 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 data refresh jobpipeline, either through a flink streaming or scheduled batch job, it is * initialized after dynamicmaterialized table is created. * </ul> * * <p>A catalog implementer can either use {@link #of(Schema, String, List, Map, Long, String, * Duration, LogicalRefreshMode, RefreshMode, RefreshHandlerRefreshStatus, 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 CatalogDynamicTableCatalogMaterializedTable extends CatalogBaseTable { @Override default TableKind getTableKind() { return TableKind.DYNAMIC_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 CatalogDynamicTableCatalogMaterializedTable} with given table options {@code options}. * * @return a new copy of this table with replaced table options */ CatalogDynamicTableCatalogMaterializedTable copy(Map<String, String> options); /** Return the snapshot specified for* theReturns table.a Return Optional.empty() if not specified. */ Optional<Long> getSnapshot(); /**copy of this {@code CatalogDynamicTable} with given refresh info. * * The@return definitiona querynew textcopy of dynamicthis table, textwith isreplaced expandedrefresh ininfo contrast to the original SQL.*/ CatalogDynamicTable *copy( This is needed because the context such as current DB is lostRefreshStatus afterrefreshStatus, 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").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, * @return the dynamic table definition in expanded textis gone. Expanded query text takes care of this, as an example. */ String getDefinitionQuery(); /** Get the freshness of * <p>For example, for a dynamic table whichthat is useddefined toin determinethe thecontext physicalof refresh mode. */"default" database with Duration getFreshness(); /** Get the logical refresh mode of dynamic table. */ * a query {@code select * from test1}, the expanded query text might become {@code select LogicalRefreshMode getLogicalRefreshMode(); /** Get the physical refresh mode of dynamic table. */ RefreshMode getRefreshMode(); * `test1`.`name`, `test1`.`value` from `default`.`test1`}, where table test1 resides in * database "default" and has two columns ("name" and "value"). * / ** Get@return the materialized refreshtable statusdefinition ofin dynamicexpanded tabletext. */ RefreshStatusString getRefreshStatusgetDefinitionQuery(); /** ReturnGet summarythe descriptionfreshness of materialized table which is used to determine the physical refresh handlermode. */ Optional<String>Duration getRefreshHandlerDescriptiongetFreshness(); /** Get the logical refresh *mode Returnof thematerialized serializable refresh handler of dynamic table. This will not be used for describe * table. table. */ LogicalRefreshMode getLogicalRefreshMode(); /** Get the physical refresh mode of materialized table. */ byte[]RefreshMode getSerializableRefreshHandlergetRefreshMode(); /** TheGet logicalthe refresh modestatus of dynamicmaterialized table. */ RefreshStatus @PublicEvolvinggetRefreshStatus(); enum LogicalRefreshMode { /** Return summary description of refresh handler. */ Optional<String> getRefreshHandlerDescription(); /** * Return the serialized *refresh Thehandler refresh pipelineof materialized table. This will not be executed in continuous mode, corresponding to {@linkused for describe * table. */ * RefreshMode#CONTINUOUS}.byte[] getSerializedRefreshHandler(); /** The logical refresh mode of materialized table. */ @PublicEvolving enum LogicalRefreshMode CONTINUOUS,{ /** * The refresh pipeline will be executed in fullcontinuous mode, corresponding to {@link * RefreshMode#FULLRefreshMode#CONTINUOUS}. */ FULLCONTINUOUS, /** * The refresh pipeline modewill isbe determinedexecuted byin freshnessfull ofmode, dynamic table, eithercorresponding to {@link * RefreshMode#FULL} or {@link RefreshMode#CONTINUOUS}. */ AUTOMATICFULL, } /** * 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 dynamicmaterialized table. */ @PublicEvolving enum RefreshMode { CONTINUOUS, FULL } /** Background refresh pipeline status of dynamicmaterialized table. */ @PublicEvolving enum RefreshStatus { INITIALIZING, ACTIVATED, SUSPENDED } } |
...
For Catalog, we don’t need to introduce any new method. But we should highlight that the Catalog support create dynamic materialized table represents a physical storage, which can provide full historical data or incremental changelog, so developers need to override the getFactory method to find the DynamicTableFactory for planner to compile a query.
Code Block | ||||
---|---|---|---|---|
| ||||
@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 dynamicmaterialized 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 dynamicmaterialized 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(); } } |
...
We introduce the class RefreshHandler that records the meta information of the current dynamic materialized table background refresh job.
Code Block | ||
---|---|---|
| ||
/**
* 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();
} |
...
ResolvedCatalogMaterializedTable
We introduce the public class ResolvedCatalogDynamicTableResolvedCatalogMaterializedTable that represents a validated CatalogDynamicTableCatalogMaterializedTable.
Code Block | ||
---|---|---|
| ||
/** * A validated {@link CatalogDynamicTableCatalogMaterializedTable} 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 ResolvedCatalogDynamicTableResolvedCatalogMaterializedTable implements ResolvedCatalogBaseTable<CatalogDynamicTable>ResolvedCatalogBaseTable<CatalogMaterializedTable>, CatalogDynamicTableCatalogMaterializedTable { private final CatalogDynamicTableCatalogMaterializedTable origin; private final ResolvedSchema resolvedSchema; public ResolvedCatalogDynamicTableResolvedCatalogMaterializedTable(CatalogDynamicTableCatalogMaterializedTable origin, ResolvedSchema resolvedSchema) { this.origin = Preconditions.checkNotNull( origin, "Original catalog dynamicmaterialized 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 ResolvedCatalogDynamicTableResolvedCatalogMaterializedTable((CatalogDynamicTableCatalogMaterializedTable) origin.copy(), resolvedSchema); } @Override public Optional<String>CatalogMaterializedTable getDescriptioncopy() { return origin.getDescription(); RefreshStatus }refreshStatus, @Override public Optional<String> getDetailedDescription() {String refreshHandlerDescription, return origin.getDetailedDescription(); byte[] serializedRefreshHandler) } { @Override publicreturn booleannew isPartitionedResolvedCatalogMaterializedTable() { return origin.isPartitioned( origin.copy(refreshStatus, refreshHandlerDescription, serializedRefreshHandler), resolvedSchema); } @Override @Override public List<String>Optional<String> getPartitionKeysgetDescription() { return origin.getPartitionKeysgetDescription(); } @Override public ResolvedCatalogDynamicTableOptional<String> copy(Map<String, String> optionsgetDetailedDescription() { return new ResolvedCatalogDynamicTable(origin.copygetDetailedDescription(options), resolvedSchema); } @Override public Optional<Long>boolean getSnapshotisPartitioned() { return Optionalorigin.emptyisPartitioned(); } @Override public CatalogDynamicTableList<String> getOrigingetPartitionKeys() { return origin.getPartitionKeys(); } @Override public ResolvedSchemaResolvedCatalogMaterializedTable getResolvedSchema(copy(Map<String, String> options) { return new ResolvedCatalogMaterializedTable(origin.copy(options), resolvedSchema); } @Override public StringOptional<Long> getDefinitionQuerygetSnapshot() { return originOptional.getDefinitionQueryempty(); } @Override public DurationCatalogMaterializedTable getFreshnessgetOrigin() { return origin.getFreshness(); } @Override public ResolvedSchema getResolvedSchema() { return resolvedSchema; } @Override public LogicalRefreshModeString getLogicalRefreshModegetDefinitionQuery() { return origin.getLogicalRefreshModegetDefinitionQuery(); } @Override public RefreshModeDuration getRefreshModegetFreshness() { return origin.getRefreshModegetFreshness(); } @Override public RefreshStatusLogicalRefreshMode getRefreshStatusgetLogicalRefreshMode() { return origin.getRefreshStatusgetLogicalRefreshMode(); } @Override public Optional<String>RefreshMode getRefreshHandlerDescriptiongetRefreshMode() { return origin.getRefreshHandlerDescriptiongetRefreshMode(); } @Nullable @Override public byte[]RefreshStatus getSerializableRefreshHandlergetRefreshStatus() { return origin.getSerializableRefreshHandlergetRefreshStatus(); } } |
Configuration
@Override
public Optional<String> getRefreshHandlerDescription() {
return origin.getRefreshHandlerDescription();
}
@Nullable
@Override
public byte[] getSerializedRefreshHandler() {
return origin.getSerializedRefreshHandler();
}
} |
Configuration
We need to introduce two We need to introduce two options to control the background refresh job.
Key | Required | Default | Type | Description |
dynamic.materialized-table.refresh-mode.freshness-threshold | Optional | 30 minute | Duration | This is used to derive the refresh mode for the background refresh job that Dynamic Materialized Table statement generates. If FRESHNESS is less than the threshold, it is running in continuous mode. If it is greater, running in full mode. |
partition.fields.#.date-formatter | Optional | None | String | Specify the time partition formatter of partitioned dynamic materialized table, # represents the string type partition field. This is used to hint to the framework which partition to refresh in full refresh mode. It is a common materialize table option that correspoding Catalog should support. |
In In the Overall Architecture section, we will introduce the SQL gateway that is responsible for managing the background refresh job of the dynamic materialized table, so this option is the only scope to SQL gateway. It is read from the flink-conf each time the service starts, and is not allowed to be modified. After the service starts, it first looks up all the Dynamic Materialized Tables, compares dynamic.materialized-table.refresh-mode.freshness-threshold value to each Dynamic Materialized Table freshness, and if the refresh mode changes, the original refresh job is stopped, then a new job is created, and the job information is updated to Dynamic Materialized Table.
Modifying this option may affect the refresh mode of the background refresh jobs of all registered dynamic materialized tables in the Catalog, so please modify it carefully. If you really need to modify the job refresh mode, please prioritize to realize it by modifying dynamic materialized table freshness.
Gateway Rest API
v3/dynamicmaterialized-tables/refresh | |
Verb: POST | Response code: 200 OK |
Trigger a refresh operation of dynamic materialized tables. | |
Request body | {[ "tablesmaterializedTable": []"", "isPeriodic": true "scheduleTime": "", "configuration" {}}scheduleTimeFormat": "", "config": { "key": "value" } ] |
Response body | { jobId jobID: "", clusterInfo: {} } |
This API is used for workflow scheduler to trigger a refresh operation of dynamic materialized table aumatically.
Proposed Design
This is just a draft API, the main purpose of which is to give you an idea of the mechanics of the refresh operation. We will give a detailed design in another FLIP that supports pluggable Workflow Scheduler.
Proposed Design
Integrating the aforementioned context, Materialized Table is a new table type introduced Integrating the aforementioned context, Dynamic Table is a new table type introduced for Flink SQL that enables users to build and manage data pipelines, it is automatically refreshed at specified freshness intervals. This means that the dynamic materialized table needs to be bound to a transformation SQL statement, and based on the defined data freshness requirements, a background data refresh job piepline will be created to automatically update the data.
So dynamic table materialized table is composed of three parts:
- Metadata: This includes the table's schema, options, definition query, freshness and refresh job piepline information.
- Data: This corresponds to a specific storage that holds the current table's data, which can provide full historical data or incremental changelog.
- Background Refresh JobPipeline: This job ensures the production and updating of data in either a continuous or full refresh waymode.
Overall Architecture
Taking into account the functional design of Dynamic Materialized Table, the management of its entire lifecycle is primarily supported by the following core components in the architecture:
- SQL CLI: A user interaction interface responsible for submitting dynamic materialized table statements.
- SQL Gateway: In charge of parsing dynamic materialized table statements, creating tables, submitting jobs to flink cluster. Additionally, the workflow scheduler will interact with SQL Gateway to submit batch workflows.
- Dynamic Materialized Table Manager: As an embedded manager in SQL Gateway that interacts with Workflow Scheduler through Restful APIs. It is responsible for deducing the data refresh mode for dynamic materialized table, deriving streaming job runtime parameters, CRON expressions for batch scheduling jobs, creating batch-scheduled jobs to scheduler, and more.
- Catalog: Storage for dynamic table metadata, which includes schema, options, freshness, definition query, refresh job information, etc.
- Pluggable Workflow Scheduler: Interacts bidirectionally with SQL Gateway through Restful APIs, responsible for creating batch-scheduled jobs and triggering their execution. It supports open-source workflow schedulers like Airflow, DolphinScheduler through a pluggable mechanism.
- Flink Cluster: Runs Flink Jobs, including both streaming and batch jobs.
...
As illustrated in the architecture diagram above, the workflow for executing a Dynamic Materialized Table create statement is as follows:
- Parse the create statement, deduce the schema of the dynamic table based on the select query, and generate a CatalogDynamicTableCatalogMaterializedTable instance, including schema, table options(if specified), definition query and freshness.
- Create the dynamic materialized table in Catalog.
- Determine the refresh mode based on defined freshness and dynamic.materialized-table.refresh-mode.freshness-threshold value, generating the corresponding type background refresh job.
- If it is the continuous refreshmode:
- Compile and submit a Flink streaming job, which starts to run statelessly, the source table data read position defaults to the entire dataset. Then the refresh job handler information is updated to CatalogDynamicTableCatalogMaterializedTable.
- If it is the full refreshmode:
- Deduce the scheduling CRONexpression via freshness value, create a scheduling workflow in the scheduler, and then update the refresh job information to CatalogDynamicTableCatalogMaterializedTable.
- The scheduler periodically triggers the batch scheduling tasks, then call the SQL gateway to execute the workflow.
- When SQL gateway receives a scheduled task, it retrieves the definition query of dynamic materialized table from Catalog, calculates the corresponding time partition value (e.g., ds) for partitioned table, then submits a Flink batch job.
Among these, the most critical component is the Dynamic Materialized Table Manager, which handles the most essential functions in the execution process of Dynamic Materialized Table and interacts with other relevant components. Since the implementation details of Dynamic Materialized Table Manager are not the focus of this FLIP, they are not elaborated on here.
...
Code Block | ||
---|---|---|
| ||
CREATE DYNAMICMATERIALIZED TABLE dt_sink PARTITIONED BY (ds) WITH( --hint time partition format 'partition.fields.ds.date-formatter' = 'yyyy-MM-dd' ) FRESHNESS = INTERVAL '1' DAY AS SELECT ds id, order_number, user_id, ... FROM source |
...
Why not use Materialized View? A view is an equivalent of a query result. According to the definition of the SQL standard, it is not supposed to support modifications. However, in the lakehouse scenario, there is a need to modify data, which leads to a conceptual conflict between the two. Therefore, using a table is more appropriate.
Base class of
...
CatalogMaterializedTable
Regarding CatalogDynamicTableCatalogMaterializedTable, another alternative is extending CatalogTable directly instead of CatalogBaseTable. This makes it straightforward to have attributes of CatalogTable.
Code Block | ||
---|---|---|
| ||
@PublicEvolving public interface CatalogMaterializedTable extends CatalogTable { /** * 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 materialized 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 the materialized table which is used to determine the data 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 }@PublicEvolving public interface CatalogDynamicTable extends CatalogTable { /** * The definition query text of dynamic 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 dynamic table definition in expanded text. */ String getDefinitionQuery(); /** Get the freshness of the dynamic table which is used to determine the data refresh mode. */ Duration getFreshness(); /** GetThe thephysical refresh mode of thematerialized dynamic table. */ @PublicEvolving enum RefreshMode getRefreshMode(); { /** Get the refresh handler of the dynamic table. */ CONTINUOUS, FULL RefreshHandler getRefreshHandler();} /** TheBackground refresh pipeline modestatus of thematerialized dynamic table. */ @PublicEvolving enum RefreshModeRefreshStatus { CONTINUOUSINITIALIZING, FULLACTIVATED, SUSPENDED } } |
There are two main reasons why CatalogTable was not inherited:
- Dynamic Materialized Table is a new conceptual entity and table type that also includes attributes such as query, freshness and refresh job, which makes it more appropriate to use the new interface.
- If CatalogTable adds a new attribute that is not supported by Dynamic Materialized Table, it will lead to a conflict, so it is more appropriate to represent it using an interface that is parallel to CatalogTable.
...
Temporary View Limitation
Since both the dynamic materialized table and the query it references need to be persistent, and the temporary view is executed locally and only effective for the current execution, it cannot be persisted. Therefore, it is not supported. If there is a real need to use temporary view, there are currently two solutions. One is to use a View, and the other is to use a Common Table Expression (CTE)[7], which is equivalent in effect to a temporary view. Of course, future support for temporary view can be determined based on user demand and in conjunction with technical implementation.
Code Block | ||
---|---|---|
| ||
CREATE DYNAMICMATERIALIZED 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 |
...
From the standpoint of strict semantics, the current design of the Dynamic Materialized Table does not support altering the query statement. In the context of a lakehouse, a common scenario involves the need to add fields due to schema evolution, which is a legitimate requirement that we intend to support in the future. One possible solution we have conceived is to use the syntax Alter Dynamic Materialized Table xxx ADD query_statement to provide this support. At the framework level, we would optimize by merging multiple query statements that write to the same Dynamic Materialized Table.
Code Block | ||
---|---|---|
| ||
-- create a dynamicmaterialized table with two agg indices CREATE DYNAMICMATERIALIZED 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 DYNAMICMATERIALIZED 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; |
...