Dataphin Flink SQL lexical enhancement-Dataphin(Dataphin)-阿里云帮助中心

Dataphin extends Flink SQL with ANSI-SQL-aligned syntax enhancements that eliminate repetitive DDL declarations and enable cross-project references for tables and functions.

Support cross-project reference

DDL and DML table name syntax enhancement

A standard Flink SQL task requires both DDL and DML statements:

--DDL
CREATE TABLE input_table (
  	field1 VARCHAR,
    field2 BIGINT
  ) WITH (
		type = 'datagen'
  );
CREATE TABLE output_table (
  	field1 VARCHAR,
    field2 BIGINT
  ) WITH (
		type = 'print'
  );
--DML
INSERT INTO output_table
SELECT field1, field2 FROM input_table;

Repeating DDL definitions across tasks is error-prone. Dataphin lets you define a schema once and reference it in any task, including tasks in other projects. The enhanced table name syntax is:

<compoundTableName> ::= [projectName <DOT>] tableName

Resolution rules:

${project_name}.tableName — references a table in the specified project
tableName — references a table in the current project (default)

Replace the full DDL block with a single DML statement that references the schema directly:

INSERT INTO output_table -- Default project prefix is the current project.
SELECT field1, field2 FROM ${project_name}.input_table;

Built-in function and UDX syntax enhancement

Standard Flink SQL requires declaring each user-defined extension (UDX) before use:

CREATE FUNCTION MY_UDX AS 'package.class';
...
CREATE VIEW my_view AS 
SELECT MY_UDX(args)
FROM ...

Dataphin has introduced the concept of resources. Once a function is registered in Dataphin, it can be directly referenced in any task — including tasks in other projects — without re-declaring it. The enhanced function name syntax is:

<compoundFunctionName> ::= [projectName <COLON>] functionName

Resolution rules:

${project_name}:functionName — references a function in the specified project
functionName — references a function in the current project (default)

Use the function directly in the development task after registering it:

CREATE VIEW my_view AS SELECT ${project_name}:MY_UDX(args) -- Default project prefix is the current project.
FROM ...

Overloading and overriding built-in functions

Dataphin lets you overload or override built-in functions. The custom version always takes precedence.

With a project prefix

SELECT ${project_name}:SUBSTRING('20200108171740', 1, 8) FROM dual;
--Dataphin will default to using the overloaded built-in function `SUBSTRING`. If the custom function is not found under `${project_name}`, it will fail during precompilation.

Without a project prefix

SELECT SUBSTRING('20200108171740', 1, 8) FROM dual;
--Dataphin will prioritize finding the custom `SUBSTRING` function in the current project. If not found, it is considered a built-in function.

Use UpdateStatement and TableHints to configure task-level DDL parameters

Dataphin schemas define shared DDL WITH parameters for reuse across tasks. For parameters that vary per task — such as WATERMARK, computed columns, primary keys, and HEADER — configure them at the task level using two complementary mechanisms. Additionally, Dataphin has introduced SetOption syntax support.

SET statement: applies a parameter globally across the entire task
TableHints: overrides a parameter for a specific query context within the task

SET parameters act as task-wide defaults. TableHints override SET for the specific query where they appear. When both are present, TableHints take precedence for that query; SET fills in any parameters not covered by TableHints.

The SET syntax is:

SET {
	[ projectName <DOT>] tableName <DOT> 
		{ 
			propertyName
			| WATERMARK
			| <computedColumn> 
			| <primaryKey> 
			| <procTime> 
			| {fieldName <DOT> <isHeader>} 
		}
} 	<EQ> {
		@variable = identifier
		| @variable = literal
		| @variable = expression
}

Separate multiple SET statements with semicolons. Define shared DDL WITH parameters in the schema; use SET or TableHints only for task-specific values.

Important

When using Flink SQL reserved words, add backtick ( ) symbols. For example:

Watermark: SET [project.]table.
Partition: SET [project.]table.

DDL parameter precedence

Task-level parameter overwrite

Task-level parameters overwrite the corresponding schema (meta table) parameters, but only for parameters that support overwriting. For example, if the schema sets the primary key of table T to field1, a task-level SET that assigns field2 as the primary key takes precedence — field1 is no longer the primary key of T within that task.

Important

Due to Dataphin's security and management policies, certain schema parameters cannot be overwritten in SQL tasks using SET. For example, the table name defined in a MaxCompute schema cannot be altered in an SQL task using SET. Overwritable parameters vary by schema type.

Task internal SET parameters

SET acts as a global default for the entire task; TableHints are context-specific. When a parameter appears in both SET and TableHints, TableHints win for that query. Parameters in SET that are absent from TableHints are applied as defaults.

Starting with Blink 2.2.5, dimension tables support INDEX and UNIQUE INDEX in addition to PRIMARY KEY for declaring JOIN KEY. Dataphin evaluates all three together. The following example shows how SET and TableHints interact:

SET my_dim_table.primarykey = 'field1';

CREATE VIEW tmp1 AS 
SELECT my_source_table1.*
FROM my_source_table1 JOIN my_dim_table WITH (index = 'field2') FOR SYSTEM_TIME AS OF PROCTIME()
ON ...;

CREATE VIEW tmp2 AS 
SELECT my_source_table2.*
FROM my_source_table2 JOIN my_dim_table WITH (primarykey = 'field3') FOR SYSTEM_TIME AS OF PROCTIME()
ON ...;

In this example:

For tmp1: SET supplies field1 as the primary key of my_dim_table (parameter padding) for the creation of tmp1, and the TableHint adds field2 as the index.
For tmp2: the TableHint overrides SET, making field3 the primary key of my_dim_table (parameter supersession).

Additional examples

Override Kafka consumer group and scan offset at task level

Given a Kafka source table:

-- vvr syntax
CREATE TEMPORARY TABLE kafkaTable (
    `user_id` BIGINT,
    `item_id` BIGINT,
    `category_id` BIGINT,
    `behavior` STRING,
    `topic` STRING METADATA VIRTUAL,
    `partition` BIGINT METADATA VIRTUAL
) WITH (
    'connector' = 'kafka',
    'topic' = 'my_excellent_topic',
    'properties.bootstrap.servers' = 'mykafka:9092',
    'properties.group.id' = 'my_excellent_group'
    'format' = 'csv',
    'scan.startup.mode' = 'earliest-offset'
)

Override the consumer group and scan offset at the task level:

-- If the table property name is a SQL keyword or contains special characters, escape it with backticks.
SET kafkaTable.`properties.group.id` = 'new_group_id';
SET kafkaTable.`scan.startup.mode` = 'latest-offset';

Override buffer size for an HBase sink table at task level

Given an HBase result table:

create table hbase_output(
  rk  varchar,
  rk1 varchar,
  rk2 varchar,
  f bigint,
  PRIMARY KEY(rk)
) with (
  type='alihbase',
  diamondKey='xxxxxxx',
  diamondGroup='yyyyyyy',
  columnFamily='cf',
  tableName='blink_hbase_test',
  bufferSize=500;
);

Override the buffer size at the task level:

SET hbase_output.bufferSize = 1000;

Set cache strategy for a MaxCompute dimension table at task level

Given a MaxCompute dimension table:

 CREATE TABLE white_list (
  id varchar,
  name varchar,
  age int,
  PRIMARY KEY (id)
  --PERIOD FOR SYSTEM_TIME -- Note: From blink3.x, dimension table DDL does not require PERIOD FOR SYSTEM_TIME.
) with (
  type = 'odps',
  endPoint = 'your_end_point_name',
  project = 'your_project_name',
  tableName = 'your_table_name',
  accessId = 'your_access_id',
  accessKey = 'your_access_key',
  `partition` = 'ds=20180905',
  cache = 'ALL'
);

Set cache strategy and TTL at the task level:

SET white_list.cache='ALL';
SET white_list.cacheTTLMs=86400000;
--If you want to specifically set the cache update time of this table at the task level, you can add SET here.

Configure WATERMARK, computed columns, HEADER, PROCTIME, and join keys

The schema supports only primary key and HEADER settings. Configure WATERMARK, PROCTIME, computed columns, and join key indexes at the task level using the following syntax:

Set WATERMARK

Syntax:

SET {
	[projectName <DOT>] tableName <DOT> WATERMARK
} 	<EQ> {
		WATERMARK [watermarkName] FOR <rowtime_field> AS withOffset(<rowtime_field>, offset)
}

Standard Flink SQL DDL with WATERMARK:

CREATE TABLE sls_stream(
		a INT,
		b BIGINT,
		c VARCHAR,
		ts TIMESTAMP,
		WATERMARK FOR ts AS withOffset(ts, 1000)
	  ) with (
		type = 'sls',
  ...);

Equivalent Dataphin SET syntax:

-- Version 3.7 and above: single quotes are required.
SET sls_stream.`watermark`= 'WATERMARK FOR ts AS withOffset(ts, 1000)';
-- Version 3.6:
SET sls_stream.`watermark`= WATERMARK FOR ts AS withOffset(ts, 1000);

Set computed columns

Syntax:

SET {
	[projectName <DOT>] tableName <DOT> <computedColumn>
} 	<EQ> {
		column_name AS computed_column_expression
    |
    {
    	<LPAREN>
    		column_name AS computed_column_expression
    		(<COMMA> column_name AS computed_column_expression)*
      <RPAREN>  
    }
}

Standard Flink SQL DDL with a computed column:

CREATE TABLE sls_stream(
		a INT,
		b BIGINT,
		c VARCHAR,
		ts AS to_timestamp(c, 'yyyyMMddHHmmss')
	  ) with (
		type = 'sls',
  ...);

Equivalent Dataphin SET syntax:

Single computed column

SET sls_stream.computedColumn= ts AS to_timestamp(c, 'yyyyMMddHHmmss');

Multiple computed columns

SET sls_stream.computedColumn= ( ts1 AS to_timestamp(c, 'yyyyMMddHHmmss'), ts2 AS to_timestamp(d, 'yyyy-MM-dd HH:mm:ss') );

Set PROCTIME

Syntax:

SET {
	[projectName <DOT>] tableName <DOT> <procTime>
} 	<EQ> {
		columnName AS PROCTIME()
}

Equivalent Dataphin SET syntax:

SET sls_stream.procTime= d AS PROCTIME();

Set PRIMARY KEY, INDEX, and UNIQUE INDEX for dimension table joins

Important

Starting with Blink 3.6.0, TableHints use the community-accepted Hints method. Neither Ververica Flink nor open-source Flink supports INDEX or UNIQUE INDEX.

TableHints syntax:

query :
      select /*+ hint_content */
      ...
      from
          table_name1 /*+ hint_content */
          join 
          table_name2 /*+ hint_content */
      ...
hint_content :
       hint_item[, hint_item]*
hint_item :
       hint_name
   |   hint_name(k1=v1 [ , k2=v2 ]*)
   |   hint_name(hint_opt [ ,hint_opt ]*)
k :
      simple_identifier
v :
      string_literal
hint_opt :
      simple_identifier
   |  numeric_literal
   |  string_literal

Basic TableHints usage for a dimension table join:

INSERT INTO table
SELECT source_table.*
FROM source_table JOIN dim_table /*+ primarykey(field1) */ FOR SYSTEM_TIME AS OF PROCTIME()
ON ...;

SET applies the JOIN KEY globally; TableHints are context-specific. The following example shows how each INSERT resolves its effective join key:

SET dim_table.index = 'field1'; -- set pk and set unique index follow the same pattern

-- Equivalent to declaring INDEX (field1) in dim_table's DDL
INSERT INTO table1
SELECT source_table.*
FROM source_table JOIN dim_table FOR SYSTEM_TIME AS OF PROCTIME()
ON ...;

-- Equivalent to declaring INDEX (field2) in dim_table's DDL.
INSERT INTO table2
SELECT source_table.*
FROM source_table JOIN dim_table WITH (index = 'field2') FOR SYSTEM_TIME AS OF PROCTIME()
ON ...;

-- Equivalent to declaring PRIMARY KEY (field1), INDEX (field2) in dim_table's DDL.
INSERT INTO table3
SELECT source_table.*
FROM source_table JOIN dim_table WITH (primarykey = 'field2') FOR SYSTEM_TIME AS OF PROCTIME()
ON ...;

-- Equivalent to declaring PRIMARY KEY (field1), INDEX (field1) in dim_table's DDL.
INSERT INTO table4
SELECT source_table.*
FROM source_table JOIN dim_table WITH (primarykey = 'field1') FOR SYSTEM_TIME AS OF PROCTIME()
ON ...;

-- Equivalent to declaring PRIMARY KEY (field1), UNIQUE INDEX (field2, field3) in dim_table's DDL.
INSERT INTO table5
SELECT source_table.*
FROM source_table JOIN dim_table WITH (UNIQUEINDEX = 'field2, field3') FOR SYSTEM_TIME AS OF PROCTIME()
ON ...;

Important

The join with syntax was used in earlier versions of the Blink engine. For Blink (newer versions), Ververica Flink, and open-source Flink, use TableHints instead.

TableHints usage across Blink, Ververica Flink, and Apache Flink:

-- Specify that the primary key of dim_table is the id field (supported by blink, vvr, apache-flink).
INSERT INTO table1
SELECT source_table.*
FROM source_table 
JOIN dim_table/*+primarykey(id)*/ FOR SYSTEM_TIME AS OF PROCTIME()
ON ...;

-- Specify that the index of dim_table is id (only supported by blink).
INSERT INTO table1
SELECT source_table.*
FROM source_table 
JOIN dim_table/*+index(id)*/ FOR SYSTEM_TIME AS OF PROCTIME()
ON ...;

-- Specify that the unique index of dim_table is id (only supported by blink).
INSERT INTO table1
SELECT source_table.*
FROM source_table 
JOIN dim_table/*+uniqueindex(id)*/ FOR SYSTEM_TIME AS OF PROCTIME()
ON ...;

Configure HEADER

Syntax:

SET {
	[projectName <DOT>] tableName <DOT> <isHeader>
} 	<EQ> {
		TRUE
}

Given a Kafka table with a HEADER field:

-- blink demo
create table kafka_table (
    `message` VARBINARY,
    topic varchar HEADER
) WITH (
    type = 'kafka010',
    topic = 'test_kafka_topic',
    `group.id` = 'test_kafka_consumer_group',
    bootstrap.servers = 'ip1:port1,ip2:port2,ip3:port3'
);

-- vvr demo
create table kafka_table (
    `message` VARBINARY,
    topic varchar METADATA VIRTUAL
) WITH (
    'connector' = 'kafka',
    'topic' = 'my_excellent_topic',
    'properties.bootstrap.servers' = 'mykafka:9092',
    'properties.group.id' = 'my_excellent_group'
    'format' = 'csv',
    'scan.startup.mode' = 'earliest-offset'
);

Mark a field as HEADER using either form:

SET kafka_table.topic.isHeader = true; -- Without single quotes.

SET kafka_table.topic.isHeader = 'true'; -- With single quotes.

Configure AccessKey for TimeTunnel auto-subscription

Configure AccessKey credentials for TimeTunnel auto-subscription:
```
set xx_project.xx_table.accessId='xxxxxx';
set xx_project.xx_table.accessKey='xxxxxx';
```
- Superuser accounts and accounts lacking permissions must manually configure subscriptions using SQL; they cannot use the automatic subscription feature.
- For other personal accounts, configuring the SET information is optional. Typically, if JobName is not specified, the default JobName default is used to record the subscription for the Topic. To prevent redundant subscriptions for the same Topic across different jobs, the subscription is not repeated within a separate precompilation function once an archive record exists. The submit and publish functions automatically subscribe to the Topic the first time they are used for a given job, and do not subscribe again if an archive record exists.
Resolve table name conflicts between offline and streaming tables

When a project is linked to an offline computing source, offline and streaming tables may share the same name. Dataphin raises a precompilation error and requires explicit disambiguation:
1. If a table name exists in both streaming and offline contexts within the same task, precompilation fails with: Please assign a single type to the table adi_ae_matrix_rt_slr_prim_cate using the syntax SET XXX.XXX.tableType = 'XX'. Available types include: [CLOUD_HBASE, MAX_COMPUTE].
2. Explicitly set the table type within the same project based on the error message. For example, to force the MaxCompute physical table:
```
-- Force the use of MaxCompute physical table.
SET xx_project.xx_table.tableType = 'MAX_COMPUTE';
```
Specify data source type for MaxCompute tables

To use a MaxCompute data source table, set tableType='odps'; to avoid a parameter missing error.
Enable column pruning for a dimension table

To manually enable column pruning, use:
```
SET {project}.{dimtable}.dataphinColumnPruning='true';
```
Note
Column pruning applies only to dimension tables and supports only direct join tables, not subqueries within direct joins.
Configure stream and batch parameters for stream-batch integration

In stream-batch integration tasks, mirror tables are commonly utilized. These mirror tables are converted into the appropriate stream or batch tables for use.

Because stream and batch tables may have different data sources and require different key parameters in the with clause — and may have unique settings like batchSize — use TableHints to map each mirror table to its corresponding stream or batch table. The syntax is:
```
set project.table.${mode}.${key}
```
Set the start and end times for a batch task:
```
set project.table.batch.startTime='2020-11-11 00:00:00';
set project.table.batch.endTime='2020-11-12 00:00:00';
```
Set the AccessKey for a stream table:
```
set dwd_lux_trd_ord_subpay_mirror.`stream`.accessId='xxxxxx';
set dwd_lux_trd_ord_subpay_mirror.`stream`.accessKey='xxxxxx';
```

Crop output columns for an HBase sink table

Use the hint method to specify which columns to write to an HBase sink table, bypassing a field-count mismatch error.

Important

This feature applies only to HBase sink tables.

-- Suppose the HBase DDL statement is as follows
CREATE TABLE hbase_sink(
  rowkey INT,
  family1 ROW<q1 INT>,
  family2 ROW<q2 STRING, q3 BIGINT>
) with (
  'connector'='cloudhbase',
  'table-name'='<yourTableName>',
  'zookeeper.quorum'='<yourZookeeperQuorum>'
);

-- Error: Because the table fields do not match
insert into hbase_sink select key, ROW(f1q1) from ...

-- Pass: Here, the hint method specifies cropping the sink table, specifying writing to rowkey and q1 columns
insert into hbase_sink/*+dataphincolumns(rowkey,family1.q1)*/
select key, ROW(f1q1) from ...

-- Pass: Here, the hint method specifies cropping the sink table, specifying writing to rowkey and q2, q3 columns
insert into hbase_sink/*+dataphincolumns(rowkey,family2.q2,family2.q3)*/
select key, ROW(f2q2,f2q3) from ...