# 做源表
#### 语法格式
用DWS-Connector做源表时，DWS-Connector实现了SupportsLimitPushDown和SupportsFilterPushDown接口，支持将limit和where条件下推到数据库执行。
```
create table dwsSource (
  attr_name attr_type 
  (',' attr_name attr_type)* 
  (','PRIMARY KEY (attr_name, ...) NOT ENFORCED)
  (',' watermark for rowtime_column_name as watermark-strategy_expression)
)
with (
  'connector' = 'dws',
  'url' = '',
  'tableName' = '',
  'username' = '',
  'password' = ''
);
```
#### 参数说明
表1数据库配置 
| 参数        | 说明                                                                  | 默认值 |
|:---|:---|:---|
| connector | flink框架区分connector参数，固定为dws。                                        | -   |
| url       | 数据库连接地址。                                                            | -   |
| username  | 配置连接用户。                                                             | -   |
| password  | 配置密码。                                                               | -   |
| tableName | 对应dws表，默认会取public模式下的表，如果非public模式则需要显式指定，格式为：**schema.tableName**。 | -   |
   
表2查询参数 
| 参数             | 说明                                        | 默认值  |
|:---|:---|:---|
| fetchSize      | jdbc statement中fetchSize参数，用于控制查询数据库返回条数。 | 1000 |
| enablePushDown | 开启条件下推：开启后limit和where条件会下推到数据库执行。         | true |
   
#### 示例
该示例是从DWS数据源中读取数据，并写入到Print结果表中，其具体步骤参考如下：
1. 在DWS中创建相应的表，表名为**dws_order** ，SQL语句参考如下：
   ```
   create table public.dws_order(
   order_id VARCHAR,
   order_channel VARCHAR,
   order_time VARCHAR,
   pay_amount FLOAT8,
   real_pay FLOAT8,
   pay_time VARCHAR,
   user_id VARCHAR,
   user_name VARCHAR,
   area_id VARCHAR);
   ```
   
2. 在DWS中执行以下SQL语句，向**dws_order** 表中插入数据：
   ```
   insert into public.dws_order
     (order_id,
     order_channel,
     order_time,
     pay_amount,
     real_pay,
     pay_time,
     user_id,
     user_name,
     area_id) values
     ('202103241000000001', 'webShop', '2021-03-24 10:00:00', '100.00', '100.00', '2021-03-24 10:02:03', '0001', 'Alice', '330106'),
     ('202103251202020001', 'miniAppShop', '2021-03-25 12:02:02', '60.00', '60.00', '2021-03-25 12:03:00', '0002', 'Bob', '330110');
   ```
   
3. 执行Flink SQL：
   ```
   CREATE TABLE dwsSource (
     order_id string,
     order_channel string,
     order_time string,
     pay_amount double,
     real_pay double,
     pay_time string,
     user_id string,
     user_name string,
     area_id string
   ) WITH (
     'connector' = 'dws',
     'url' = 'jdbc:gaussdb://DWSIP:DWSPort/DWSdbName',
     'tableName' = 'dws_order',
     'username' = 'DWSUserName',
     'password' = 'password'
   );
   CREATE TABLE printSink (
     order_id string,
     order_channel string,
     order_time string,
     pay_amount double,
     real_pay double,
     pay_time string,
     user_id string,
     user_name string,
     area_id string
   ) WITH (
     'connector' = 'print'
   );
   insert into printSink select * from dwsSource;
   ```
   
4. 执行结果如下： ![](https://support.huaweicloud.com/tg-dws/figure/zh-cn_image_0000001910080986.png "点击放大")
   
 
#### 常见问题
- Q：我想新建一个忽略尾空的数据库表，并且通过Flink SQL筛选数据，这种场景是否支持？ A：**支持此场景** 。但是当数据库新建时添加了**td_rtrim**参数，并且Flink SQL中WHERE子句添加了等值的筛选条件，可能导致数据不一致的问题：
  Flink默认有**列裁剪**的行为（SELECT中会去掉筛选列，直接拼接WHERE子句中条件的值），这会导致最终的结果使用了WHERE子句中的值，而不是真正源端的数据。
  假设一个场景，源端a列的值有3个值'a'、'a '、'a '，由于**td_rtrim** 参数，那么**select a from xxx where a = 'a'** 会返回3个结果，但是由于Flink**列裁剪** 的行为，会导致最终3个结果均被错误的拼成了'a'，这就导致了**数据一致性错误**。
  **规避手段：** 在Flink SQL的WHERE条件中添加**rtrim** 函数，将筛选列包住，如**select \* from xxx where rtrim(xxx) = 'a'**，即可解决上述问题。