Realtime Compute for Apache Flink:FETCH_CONTENT, TRY_FETCH_CONTENT

Engine requirement

• Only Realtime Compute engine VVR 11.7.0 and later support the FETCH_CONTENT function.

• Only Realtime Compute engine VVR 11.9.0-preview.1 and later support the TRY_FETCH_CONTENT function.

Syntax

VARBINARY FETCH_CONTENT(VARCHAR uri)
VARBINARY FETCH_CONTENT(VARCHAR uri, INTEGER concurrency)

VARBINARY TRY_FETCH_CONTENT(VARCHAR uri)
VARBINARY TRY_FETCH_CONTENT(VARCHAR uri, INTEGER concurrency)

Parameters

Return value

TRY_FETCH_CONTENT tolerates only content-fetching failures at runtime. SQL validation errors, such as a wrong number or types of arguments or a concurrency value out of range, still cause the job submission to fail.

Retry and timeout configuration

FETCH_CONTENT and TRY_FETCH_CONTENT use the unified retry and timeout configuration of async scalar functions. Request concurrency, timeout, and retries are managed by the Flink Async Scalar operator. You can change the related parameters with SET statements in your SQL job.

The following example sets the retry strategy to fixed-delay retry with a maximum of 2 retries, waits 1 s between two requests, and sets the async call timeout of a single record to 30 s.

SET 'table.exec.async-scalar.retry-strategy' = 'FIXED_DELAY';
SET 'table.exec.async-scalar.max-attempts' = '2';
SET 'table.exec.async-scalar.retry-delay' = '1 s';
SET 'table.exec.async-scalar.timeout' = '30 s';

When table.exec.async-scalar.max-attempts is set to 2, the function sends up to 3 requests: the first request plus 2 retries. If the timeout does not leave enough time for all retries, the timeout result applies: FETCH_CONTENT throws an exception and TRY_FETCH_CONTENT returns NULL.

The following table describes all parameters.

Example 1: Download file content

• Test data

  Table 1. T1

  input	uri (VARCHAR)
  1	http://example.com/image_url
  2	oss://example-bucket/example.pdf
  3	NULL

• Test query

  SELECT 
      id,
      FETCH_CONTENT(uri) AS `value`
  FROM 
      T1;

• Test result

  id (INT)	value (VARBINARY)
  1	x'ffd8ffe00010......'
  2	x'aaffd8ffe000......'
  3	NULL

Example 2: Return NULL when a download fails

• Test data

  Table 1. T1

  input	uri (VARCHAR)
  1	http://example.com/image_url
  2	oss://example-bucket/example.pdf
  3	invalid://path

• Use TRY_FETCH_CONTENT to download files. If a download fails, the content of that record returns NULL, and the job is not interrupted by the failure.

  SELECT
    id,
    TRY_FETCH_CONTENT(uri) AS content
  FROM T2;

• Test result

  id (INT)	value (VARBINARY)
  1	x'ffd8ffe00010......'
  2	NULL
  3	NULL

• You can filter out records that failed to download to avoid passing NULL to downstream multimodal inference functions.

  SELECT id, content
  FROM (
    SELECT
      id,
      TRY_FETCH_CONTENT(uri) AS content
    FROM T2
  )
  WHERE content IS NOT NULL;

Example 3: Specify the concurrency

The following example sets the content-fetching concurrency of each operator instance to 8. The 8 must be an integer literal in the SQL statement.

SELECT
  id,
  TRY_FETCH_CONTENT(uri, 8) AS content
FROM T2;

The following calls are invalid and cause errors at the SQL validation stage.

-- concurrency cannot reference a column.
SELECT TRY_FETCH_CONTENT(uri, concurrency_column) FROM T3;

-- Valid values of concurrency: 1 to 1024.
SELECT TRY_FETCH_CONTENT(uri, 0) FROM T3;
SELECT TRY_FETCH_CONTENT(uri, 1025) FROM T3;

References

• Built-in functions: Supported functions.

• User-defined functions: UDFs and Manage user-defined functions (UDFs).