https://flink.apache.org/ logo
Join Slack
Powered by
Is there a way in PyFlink I can specify a `DEFAULT...
# troubleshooting
j
Is there a way in PyFlink I can specify a 
DEFAULT
value for a column? If my original table is as follows:
Copy code
return """CREATE TABLE {0} (
        ticker VARCHAR(6),
        {2} TIMESTAMP(3),
        window_end TIMESTAMP(3),
        first_price DOUBLE,
        last_price DOUBLE,
        min_price DOUBLE,
        max_price DOUBLE,
        notify_email VARCHAR(320),
        WATERMARK FOR {2} AS {2} - INTERVAL '{3}' SECOND
    ) WITH (
        'connector' = '{1}'
    )""".format(
        table_name,
        connector,
        field_watermark,
        watermark_interval
    )
How can I perform something like the following:
Copy code
return """CREATE TABLE {0} (
        ticker VARCHAR(6),
        {2} TIMESTAMP(3),
        window_end TIMESTAMP(3),
        first_price DOUBLE,
        last_price DOUBLE,
        min_price DOUBLE,
        max_price DOUBLE,
        notify_email VARCHAR(320) METADATA DEFAULT '<mailto:my@email.com|my@email.com>',
        WATERMARK FOR {2} AS {2} - INTERVAL '{3}' SECOND
    ) WITH (
        'connector' = '{1}'
    )""".format(
        table_name,
        connector,
        field_watermark,
        watermark_interval
    )
When I try something like that I get the following error:
Copy code
py4j.protocol.Py4JJavaError: An error occurred while calling o4.executeSql.
: org.apache.flink.table.api.SqlParserException: SQL parse failed. Encountered "DEFAULT" at line 9, column 44.
Was expecting one of:
    "FROM" ...
    "COMMENT" ...
    "VIRTUAL" ...
    ")" ...
    "," ...
    
	at org.apache.flink.table.planner.parse.CalciteParser.parse(CalciteParser.java:56)
	at org.apache.flink.table.planner.delegation.ParserImpl.parse(ParserImpl.java:98)
	at org.apache.flink.table.api.internal.TableEnvironmentImpl.executeSql(TableEnvironmentImpl.java:724)
	at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
	at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
	at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
	at java.base/java.lang.reflect.Method.invoke(Method.java:566)
	at org.apache.flink.api.python.shaded.py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244)
	at org.apache.flink.api.python.shaded.py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:357)
	at org.apache.flink.api.python.shaded.py4j.Gateway.invoke(Gateway.java:282)
	at org.apache.flink.api.python.shaded.py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)
	at org.apache.flink.api.python.shaded.py4j.commands.CallCommand.execute(CallCommand.java:79)
	at org.apache.flink.api.python.shaded.py4j.GatewayConnection.run(GatewayConnection.java:238)
	at java.base/java.lang.Thread.run(Thread.java:829)
Caused by: org.apache.calcite.sql.parser.SqlParseException: Encountered "DEFAULT" at line 9, column 44.
Was expecting one of:
    "FROM" ...
    "COMMENT" ...
    "VIRTUAL" ...
    ")" ...
    "," ...
    
	at org.apache.flink.sql.parser.impl.FlinkSqlParserImpl.convertException(FlinkSqlParserImpl.java:452)
	at org.apache.flink.sql.parser.impl.FlinkSqlParserImpl.normalizeException(FlinkSqlParserImpl.java:215)
	at org.apache.calcite.sql.parser.SqlParser.handleException(SqlParser.java:140)
	at org.apache.calcite.sql.parser.SqlParser.parseQuery(SqlParser.java:155)
	at org.apache.calcite.sql.parser.SqlParser.parseStmt(SqlParser.java:180)
	at org.apache.flink.table.planner.parse.CalciteParser.parse(CalciteParser.java:54)
	... 13 more
Caused by: org.apache.flink.sql.parser.impl.ParseException: Encountered "DEFAULT" at line 9, column 44.
Was expecting one of:
    "FROM" ...
    "COMMENT" ...
    "VIRTUAL" ...
    ")" ...
    "," ...
    
	at org.apache.flink.sql.parser.impl.FlinkSqlParserImpl.generateParseException(FlinkSqlParserImpl.java:39897)
	at org.apache.flink.sql.parser.impl.FlinkSqlParserImpl.jj_consume_token(FlinkSqlParserImpl.java:39708)
	at org.apache.flink.sql.parser.impl.FlinkSqlParserImpl.SqlCreateTable(FlinkSqlParserImpl.java:5221)
	at org.apache.flink.sql.parser.impl.FlinkSqlParserImpl.SqlCreateExtended(FlinkSqlParserImpl.java:6239)
	at org.apache.flink.sql.parser.impl.FlinkSqlParserImpl.SqlCreate(FlinkSqlParserImpl.java:20999)
	at org.apache.flink.sql.parser.impl.FlinkSqlParserImpl.SqlStmt(FlinkSqlParserImpl.java:3421)
	at org.apache.flink.sql.parser.impl.FlinkSqlParserImpl.SqlStmtEof(FlinkSqlParserImpl.java:3924)
	at org.apache.flink.sql.parser.impl.FlinkSqlParserImpl.parseSqlStmtEof(FlinkSqlParserImpl.java:263)
	at org.apache.calcite.sql.parser.SqlParser.parseQuery(SqlParser.java:153)
	... 15 more


Process finished with exit code 1
d
@Jeff Levesque It support computed column and so you could use it as following:
Copy code
return """CREATE TABLE {0} (
        ticker VARCHAR(6),
        {2} TIMESTAMP(3),
        window_end TIMESTAMP(3),
        first_price DOUBLE,
        last_price DOUBLE,
        min_price DOUBLE,
        max_price DOUBLE,
        notify_email VARCHAR(320),
        notify_email_with_default_value if(notify_email is null, '<mailto:my@email.com|my@email.com>', notify_email),
        WATERMARK FOR {2} AS {2} - INTERVAL '{3}' SECOND
    ) WITH (
        'connector' = '{1}'
    )""".format(
        table_name,
        connector,
        field_watermark,
        watermark_interval
    )
See the 
Computed Columns
section on https://nightlies.apache.org/flink/flink-docs-release-1.16/docs/dev/table/sql/create/#columns for more details.
j
Just tried that in my PyFlink 1.13, and got the following error:
Copy code
py4j.protocol.Py4JJavaError: An error occurred while calling o4.executeSql.
: org.apache.flink.table.api.SqlParserException: SQL parse failed. Encountered "(" at line 10, column 47.
Was expecting one of:
    "CONSTRAINT" ...
    "NOT" ...
    "NULL" ...
    "PRIMARY" ...
    "UNIQUE" ...
    "COMMENT" ...
    "METADATA" ...
    ")" ...
    "," ...
    "MULTISET" ...
    "ARRAY" ...
    "." ...
    
	at org.apache.flink.table.planner.parse.CalciteParser.parse(CalciteParser.java:56)
	at org.apache.flink.table.planner.delegation.ParserImpl.parse(ParserImpl.java:98)
	at org.apache.flink.table.api.internal.TableEnvironmentImpl.executeSql(TableEnvironmentImpl.java:724)
	at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
	at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
	at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
	at java.base/java.lang.reflect.Method.invoke(Method.java:566)
	at org.apache.flink.api.python.shaded.py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244)
	at org.apache.flink.api.python.shaded.py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:357)
	at org.apache.flink.api.python.shaded.py4j.Gateway.invoke(Gateway.java:282)
	at org.apache.flink.api.python.shaded.py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)
	at org.apache.flink.api.python.shaded.py4j.commands.CallCommand.execute(CallCommand.java:79)
	at org.apache.flink.api.python.shaded.py4j.GatewayConnection.run(GatewayConnection.java:238)
	at java.base/java.lang.Thread.run(Thread.java:829)
Caused by: org.apache.calcite.sql.parser.SqlParseException: Encountered "(" at line 10, column 47.
Was expecting one of:
    "CONSTRAINT" ...
    "NOT" ...
    "NULL" ...
    "PRIMARY" ...
    "UNIQUE" ...
    "COMMENT" ...
    "METADATA" ...
    ")" ...
    "," ...
    "MULTISET" ...
    "ARRAY" ...
    "." ...
    
	at org.apache.flink.sql.parser.impl.FlinkSqlParserImpl.convertException(FlinkSqlParserImpl.java:452)
	at org.apache.flink.sql.parser.impl.FlinkSqlParserImpl.normalizeException(FlinkSqlParserImpl.java:215)
	at org.apache.calcite.sql.parser.SqlParser.handleException(SqlParser.java:140)
	at org.apache.calcite.sql.parser.SqlParser.parseQuery(SqlParser.java:155)
	at org.apache.calcite.sql.parser.SqlParser.parseStmt(SqlParser.java:180)
	at org.apache.flink.table.planner.parse.CalciteParser.parse(CalciteParser.java:54)
	... 13 more
Caused by: org.apache.flink.sql.parser.impl.ParseException: Encountered "(" at line 10, column 47.
Was expecting one of:
    "CONSTRAINT" ...
    "NOT" ...
    "NULL" ...
    "PRIMARY" ...
    "UNIQUE" ...
    "COMMENT" ...
    "METADATA" ...
    ")" ...
    "," ...
    "MULTISET" ...
    "ARRAY" ...
    "." ...
    
	at org.apache.flink.sql.parser.impl.FlinkSqlParserImpl.generateParseException(FlinkSqlParserImpl.java:39897)
	at org.apache.flink.sql.parser.impl.FlinkSqlParserImpl.jj_consume_token(FlinkSqlParserImpl.java:39708)
	at org.apache.flink.sql.parser.impl.FlinkSqlParserImpl.SqlCreateTable(FlinkSqlParserImpl.java:5221)
	at org.apache.flink.sql.parser.impl.FlinkSqlParserImpl.SqlCreateExtended(FlinkSqlParserImpl.java:6239)
	at org.apache.flink.sql.parser.impl.FlinkSqlParserImpl.SqlCreate(FlinkSqlParserImpl.java:20999)
	at org.apache.flink.sql.parser.impl.FlinkSqlParserImpl.SqlStmt(FlinkSqlParserImpl.java:3421)
	at org.apache.flink.sql.parser.impl.FlinkSqlParserImpl.SqlStmtEof(FlinkSqlParserImpl.java:3924)
	at org.apache.flink.sql.parser.impl.FlinkSqlParserImpl.parseSqlStmtEof(FlinkSqlParserImpl.java:263)
	at org.apache.calcite.sql.parser.SqlParser.parseQuery(SqlParser.java:153)
	... 15 more


Process finished with exit code 1
d
@Jeff Levesque Have you tried the latest version, e.g. PyFlink 1.16?
10 Views
Open in Slack
PreviousNext
Powered by