Realtime Compute for Apache Flink:Common SQL errors

Problem description

If a Plain Old Java Object (POJO) class is used as the data types for the return values of a user-defined table-valued function (UDTF) and the alias names of the returned fields of the UDTF are explicitly declared in the SQL statement, fields may be misaligned. In this case, the fields that are used may not meet requirements even if the data types of the fields are consistent.

For example, the SQL verification failure occurs in the following situations: You use the following POJO class as the data types for the return values of a UDTF, package the UDTF based on the requirements described in Overview, and register the UDTF based on the requirements described in Register a deployment-level UDF.

package com.aliyun.example;

public class TestPojoWithoutConstructor {
	public int c;
	public String d;
	public boolean a;
	public String b;
}

package com.aliyun.example;

import org.apache.flink.table.functions.TableFunction;

public class MyTableFuncPojoWithoutConstructor extends TableFunction<TestPojoWithoutConstructor> {
	private static final long serialVersionUID = 1L;

	public void eval(String str1, Integer i2) {
		TestPojoWithoutConstructor p = new TestPojoWithoutConstructor();
		p.d = str1 + "_d";
		p.c = i2 + 2;
		p.b = str1 + "_b";
		collect(p);
	}
}

CREATE TEMPORARY FUNCTION MyTableFuncPojoWithoutConstructor as 'com.aliyun.example.MyTableFuncPojoWithoutConstructor';

CREATE TEMPORARY TABLE src ( 
  id STRING,
  cnt INT
) WITH (
  'connector' = 'datagen'
);

CREATE TEMPORARY TABLE sink ( 
  f1 INT,
  f2 STRING,
  f3 BOOLEAN,
  f4 STRING
) WITH (
 'connector' = 'print'
);

INSERT INTO sink
SELECT T.* FROM src, LATERAL TABLE(MyTableFuncPojoWithoutConstructor(id, cnt)) AS T(c, d, a, b);

The following error message for SQL verification is reported:

org.apache.flink.table.api.ValidationException: SQL validation failed. Column types of query result and sink for 'vvp.default.sink' do not match.
Cause: Sink column 'f1' at position 0 is of type INT but expression in the query is of type BOOLEAN NOT NULL.
Hint: You will need to rewrite or cast the expression.

Query schema: [c: BOOLEAN NOT NULL, d: STRING, a: INT NOT NULL, b: STRING]
Sink schema:  [f1: INT, f2: STRING, f3: BOOLEAN, f4: STRING]
	at org.apache.flink.table.sqlserver.utils.FormatValidatorExceptionUtils.newValidationException(FormatValidatorExceptionUtils.java:41)

In this example, the fields returned from the UDTF and the fields in the POJO class may be misaligned. In the SQL statement, Field c is of the BOOLEAN data type, while Field a is of the INT data type, which is the opposite of the data types specified by the POJO class.

Cause

The order of the returned fields varies based on a parameterized constructor of the POJO class:

• If the POJO class implements a parameterized constructor, the fields are sorted based on the order of the parameters of the parameterized constructor.

• If the POJO class does not implement a parameterized constructor, the fields are sorted based on the alphabetical order of the field names.

In this example, the POJO class does not implement a parameterized constructor. As a result, the data types for the return values of the UDTF are BOOLEAN a, VARCHAR(2147483647) b, INTEGER c, VARCHAR(2147483647) d). No error occurs in the preceding example. However, a rename list LATERAL TABLE(MyTableFuncPojoWithoutConstructor(id, cnt)) AS T(c, d, a, b) is added to the returned fields in the SQL statement. The returned fields are renamed. The renaming operation is performed based on the field position. As a result, fields are misaligned when the POJO class is used. This causes verification errors or unexpected data misalignment.

Solutions

• If the POJO class does not implement a parameterized constructor, do not explicitly rename the fields returned by the UDTF. For example, you can change the SELECT clause in the preceding INSERT statement to the following SELECT clause:

  -- If the POJO class does not implement a parameterized constructor, we recommend that you select the required field names. When you use T.*, you must know the actual order of the returned fields. 
  SELECT T.c, T.d, T.a, T.b FROM src, LATERAL TABLE(MyTableFuncPojoWithoutConstructor(id, cnt)) AS T;

• Implement a parameterized constructor in the POJO class to determine the order of the returned fields. In this case, the order of the returned fields is the order of the parameters of the parameterized constructor.

  package com.aliyun.example;
  
  public class TestPojoWithConstructor {
  	public int c;
  	public String d;
  	public boolean a;
  	public String b;
  
  	// Using specific fields order instead of alphabetical order
  	public TestPojoWithConstructor(int c, String d, boolean a, String b) {
  		this.c = c;
  		this.d = d;
  		this.a = a;
  		this.b = b;
  	}
  }

Problem description

The table.exec.mini-batch.size parameter is not configured for the deployment or the table.exec.mini-batch.size parameter is set to a negative value. The deployment includes both WindowAggregate and GroupAggregate operators. The time column of the WindowAggregate operator is proctime, which indicates the event time. The topology of the deployment contains the LocalGroupAggregate operator, but does not contain the MiniBatchAssigner operator when the deployment is started.

The following sample code provides an example on how to create a deployment that includes both WindowAggregate and GroupAggregate operators. The time column of the WindowAggregate operator is proctime, which indicates the event time.

CREATE TEMPORARY TABLE s1 (
  a INT,
  b INT,
  ts as PROCTIME(),
  PRIMARY KEY (a) NOT ENFORCED
) WITH (
  'connector'='datagen',
  'rows-per-second'='1',
  'fields.b.kind'='random',
  'fields.b.min'='0',
  'fields.b.max'='10'
);

CREATE TEMPORARY TABLE sink (
  a BIGINT,
  b BIGINT
) WITH (
  'connector'='print'
);

CREATE TEMPORARY VIEW window_view AS
SELECT window_start, window_end, a, sum(b) as b_sum FROM TABLE(TUMBLE(TABLE s1, DESCRIPTOR(ts), INTERVAL '2' SECONDS)) GROUP BY window_start, window_end, a;

INSERT INTO sink SELECT count(distinct a), b_sum FROM window_view GROUP BY b_sum;

Cause

The managed memory is used to cache data in miniBatch processing mode if the table.exec.mini-batch.size parameter is not configured for the deployment or the table.exec.mini-batch.size parameter is set to a negative value. The MiniBatchAssigner operator fails to be generated and cannot send the watermark message to compute operators to trigger final calculation and data output. Final calculation and data output are triggered only when one of the following conditions is met: The managed memory is full, the CHECKPOINT command is received and checkpointing has not been performed, and the deployment is canceled. For more information, see table.exec.mini-batch.size. The checkpoint interval is set to an excessively large value. The LocalGroupAggregate operator does not trigger data output for a long period of time.

Solutions

• Decrease the checkpoint interval. This way, the LocalGroupAggregate operator can automatically trigger data output before checkpointing is performed.

• Use the heap memory to cache data. This way, data output is automatically triggered when the amount of data cached on the LocalGroupAggregate operator reaches the value of the table.exec.mini-batch.size parameter. To configure the table.exec.mini-batch.size parameter, perform the following steps: In the Parameters section of the Configuration tab on the O&M > Deployments page in the development console of Realtime Compute for Apache Flink, set the table.exec.mini-batch.size parameter to a positive value N in the Other Configuration field.

Cause 1: A field of the TIMESTAMP_LTZ (TIMESTAMP(p) WITH LOCAL TIME ZONE) data type is used to declare the watermark in the source table. As a result, a time difference exists between the current time and the values of the watermark-related parameters.

The following example shows the difference between the watermark that is declared by using a field of the TIMESTAMP_LTZ data type and the watermark that is declared by using a field of the TIMESTAMP data type.

• The following sample code shows that the field that is used to declare the watermark in the source table is of the TIMESTAMP_LTZ data type.

  CREATE TEMPORARY TABLE s1 (
    a INT,
    b INT,
    ts as CURRENT_TIMESTAMP,-- Use the built-in function CURRENT_TIMESTAMP to generate data of the TIMESTAMP_LTZ data type. 
    WATERMARK FOR ts AS ts - INTERVAL '5' SECOND 
  ) WITH (
    'connector'='datagen',
    'rows-per-second'='1',
    'fields.b.kind'='random','fields.b.min'='0','fields.b.max'='10'
  );
  
  CREATE TEMPORARY TABLE t1 (
    k INT,
    ts_ltz timestamp_ltz(3),
    cnt BIGINT
  ) WITH ('connector' = 'print');
  
  -- Obtain the calculation result. 
  INSERT INTO t1
  SELECT b, window_start, COUNT(*) FROM
  TABLE(
      TUMBLE(TABLE s1, DESCRIPTOR(ts), INTERVAL '5' SECOND))
  GROUP BY b, window_start, window_end;

  Note

  The calculation result that is generated by using the syntax of the legacy window is the same as the calculation result that is generated by using the table-valued function (TVF) window. The following sample code provides an example of the syntax of the legacy window.

  SELECT b, TUMBLE_END(ts, INTERVAL '5' SECOND), COUNT(*) FROM s1 GROUP BY TUMBLE(ts, INTERVAL '5' SECOND), b;

  The following figures show that an 8-hour time difference exists between the current time (UTC+8) and the time in the values of the Low Watermark and Datetime of Watermark Timestamp parameters on the Watermarks tab of the Status tab, as well as between the current time (UTC+8) and the time in the value of the Task InputWatermark metric in the Watermark section of the Metrics tab after a draft is deployed and published in the development console of Realtime Compute for Apache Flink.

  • Watermark and Low Watermark

    

  • Task InputWatermark

    

• The following sample code shows that the field that is used to declare the watermark in the source table is of the TIMESTAMP (TIMESTAMP(p) WITHOUT TIME ZONE) data type.

  CREATE TEMPORARY TABLE s1 (
    a INT,
    b INT,
    -- No time zone information is included in the timestamp of the simulated data source. In this case, the timestamp is incremented by one second from 2024-01-31 01:00:00. 
    ts as TIMESTAMPADD(SECOND, a, TIMESTAMP '2024-01-31 01:00:00'),
    WATERMARK FOR ts AS ts - INTERVAL '5' SECOND 
  ) WITH (
    'connector'='datagen',
    'rows-per-second'='1',
    'fields.a.kind'='sequence','fields.a.start'='0','fields.a.end'='100000',
    'fields.b.kind'='random','fields.b.min'='0','fields.b.max'='10'
  );
  
  CREATE TEMPORARY TABLE t1 (
    k INT,
    ts_ltz timestamp_ltz(3),
    cnt BIGINT
  ) WITH ('connector' = 'print');
  
  -- Obtain the calculation result. 
  INSERT INTO t1
  SELECT b, window_start, COUNT(*) FROM
  TABLE(
      TUMBLE(TABLE s1, DESCRIPTOR(ts), INTERVAL '5' SECOND))
  GROUP BY b, window_start, window_end;

  After you deploy and publish a draft in the development console of Realtime Compute for Apache Flink, the time in the values of the Low Watermark and Datetime of Watermark Timestamp parameters on the Watermarks tab of the Status tab and the time in the value of the Task InputWatermark metric in the Watermark section of the Metrics tab are the same as the current time. In this example, the current time is the time of the simulated data.

  • Watermark and Low Watermark

    

  • Task InputWatermark

    

Cause 2: The time zone of the display time in the development console of Realtime Compute for Apache Flink is different from the time zone of the display time on the Apache Flink UI.

The display time in the development console of Realtime Compute for Apache Flink is in UTC+0. However, the display time on the Apache Flink UI is the local time that is converted based on the local time zone that the Apache Flink UI obtains by using the browser. The following example shows the difference between the display time in the development console of Realtime Compute for Apache Flink and the display time on the Apache Flink UI when UTC+8 is used. The display time in the development console of Realtime Compute for Apache Flink is 8 hours earlier than the display time on the Apache Flink UI.