<a id="flink-sql-batch-statements"></a>

# Run Batch Statements with Confluent Manager for Apache Flink

A batch statement runs its SQL query as an Apache Flink® batch job. The job reads a
bounded amount of input, writes its result and finishes. Use batch statements
for work that has an end, such as a backfill into a table or a report over a
fixed range of a topic. A statement runs in batch mode when
`execution.runtime-mode` is `BATCH`, set either in the statement’s
`flinkConfiguration` or in the environment’s statement defaults. When both
set it, the environment’s value wins, as described in
[How CMF decides the runtime mode](#flink-sql-batch-runtime-mode).

Batch statements have the following requirements:

* The statement must run on a compute pool that runs Flink 2.x. On a compute
  pool that runs Flink 1.x, such as 1.19, CMF rejects the create request
  with a 400 and the message `batch statements require Flink 2.x`.
* Every table the statement reads must be bounded. CMF accepts a batch
  statement that reads an unbounded source, but the statement moves to
  `FAILED` with an error that says the query requires all referenced sources
  to be bounded.
* `execution.runtime-mode` must be `BATCH` or `STREAMING`. CMF rejects
  `AUTOMATIC` with a 400, in a statement and in an environment’s statement
  defaults.

## Create a batch statement

A table backed by a Apache Kafka® topic is unbounded by default. To read it as a bounded
source, set the `scan.bounded.mode` table option, either with
`CREATE TABLE` or `ALTER TABLE`, or in a query hint as in the following
example. With `latest-offset`, the job reads each partition up to the
offset that was the latest when the job started. For the other values, see
[Table options](table-operations.md#table-options).

`scan.bounded.mode` sets only where the job stops reading. It starts reading
where the table’s `scan.startup.mode` says. To read a topic from its first
offset, whatever the table sets, also set `scan.startup.mode` to
`earliest-offset` in the hint.

The following statement copies all orders that exist when the job starts into
the `orders_archive` table:

```json
{
  "apiVersion": "cmf.confluent.io/v1",
  "kind": "Statement",
  "metadata": {
    "name": "archive-orders"
  },
  "spec": {
    "statement": "INSERT INTO orders_archive SELECT * FROM orders /*+ OPTIONS('scan.startup.mode'='earliest-offset', 'scan.bounded.mode'='latest-offset') */;",
    "properties": {
      "sql.current-catalog": "kafka-cat",
      "sql.current-database": "kafka-db"
    },
    "flinkConfiguration": {
      "execution.runtime-mode": "BATCH"
    },
    "computePoolName": "pool-flink-2"
  }
}
```

Create the statement as described in [Create Statements with Confluent Manager for Apache Flink](create-statements.md#flink-sql-statements-create). A
batch `INSERT INTO` statement or statement set does not need
`execution.checkpointing.interval`, because batch jobs do not checkpoint.

After CMF compiles the statement, `status.traits.runtimeMode` shows the
mode its job runs in, `BATCH` or `STREAMING`. Check this field rather than
the statement’s own `flinkConfiguration`, because the environment can
override it, as described in the next section.

When the job finishes, the statement moves to `COMPLETED`. For an
`INSERT INTO` statement or statement set, CMF then deletes the job’s Flink
resources, unless you turned off
[INSERT statement resource cleanup](../../installation/sql-statements.md#cmf-configure-sql). A batch
`SELECT` statement moves to `COMPLETED` when you read the last page of its
results. To run any batch statement again, create a new statement.

A batch `SELECT` statement buffers its results until you fetch them. The
buffer holds 4MiB by default, and when it is full, the job waits. For a
result larger than that, fetch the results while the statement runs, not
after it finishes. For details, see [Manage Statements in Confluent Manager for Apache Flink](manage-statements.md#flink-sql-statements-manage).

<a id="flink-sql-batch-runtime-mode"></a>

## How CMF decides the runtime mode

A statement’s runtime mode can come from two places:

* The `flinkConfiguration` of the statement.
* The environment’s statement defaults. `statementDefaults.interactive`
  applies to `SELECT` statements, and `statementDefaults.detached` applies
  to `INSERT INTO` statements and statement sets.

When both set `execution.runtime-mode`, the environment’s statement defaults
win. When neither sets it, the statement runs in streaming mode. The compute
pool’s Flink configuration and the environment’s `computePoolDefaults` do not
count: `execution.runtime-mode` set there has no effect on statements. A
`SET 'execution.runtime-mode' = 'batch'` statement fails with `Statement
type not supported yet.`, so set the runtime mode in one of the two places
above.

This means that you can make every `INSERT INTO` statement and statement set
in an environment a batch statement by setting the runtime mode in the
detached statement defaults:

```json
"statementDefaults": {
  "detached": {
    "flinkConfiguration": {
      "execution.runtime-mode": "BATCH"
    }
  }
}
```

It also means that a statement that sets `BATCH` itself runs in streaming
mode when the environment’s defaults set `STREAMING`. This holds whether the
statement’s sources are bounded or not. A streaming `INSERT INTO` statement
or statement set needs a checkpoint interval. Without one, CMF rejects the
create request with a 400 and an error that says the environment’s defaults
take precedence over the statement’s `BATCH`.

CMF compiles a statement for the runtime mode it has when you create it, and
the statement keeps that mode. If you change `execution.runtime-mode` in an
environment’s statement defaults, CMF rejects the update with a 400 while any
statement in the environment would switch between batch and streaming mode,
and the error names those statements. Stopped statements block the update
too. Statements that are `COMPLETED` or `FAILED` do not. Delete the
statements the error names, and then update the environment.

## How batch statements differ from streaming statements

### Savepoints

Flink cannot take a savepoint of a batch job or start a batch job from one.
CMF therefore rejects savepoint settings on batch statements:

* Creating a batch statement with `spec.savepointSchedule` fails with a 400.
  An update that adds a schedule fails with a 422.
* Creating or updating an environment whose `statementDefaults.detached` sets
  both `BATCH` and a `savepointSchedule` fails with a 400.
* A statement that sets `BATCH` in its own `flinkConfiguration` does not
  inherit a savepoint schedule from the environment’s detached statement
  defaults.
* A request to create a savepoint for a batch statement fails with a 400.
* An update that adds `spec.startFromSavepoint` to a batch statement fails
  with a 422.

### Stop, resume, and rescale

Stopping a batch statement cancels its job immediately. A batch job cannot
continue from where it stopped, so starting it again would run the whole
query from the beginning and could write its output a second time. For a batch
`INSERT INTO` statement or statement set, CMF prevents this:

* Resuming the statement after its job has run fails with a 400.
* An update that would restart the running job, such as a change to
  `parallelism`, fails with a 400. An update that does not touch the job,
  such as a label change, is accepted. If the environment’s defaults or the
  compute pool’s configuration changed since the job started, or CMF was
  upgraded or reconfigured, applying any update would restart the job, so a
  label change is rejected too.
* Stopping the statement is always accepted.

A statement that you create with `"stopped": true` has not run a job yet, so
you can start it once. To run the query again after its job has run, create a
new statement with another name, or delete the old statement first.

### Task failures

Flink restarts a failed task only when a restart strategy is configured or
checkpointing is on, and batch jobs do not checkpoint. CMF therefore adds a
default restart strategy to batch `INSERT INTO` statements and statement
sets: `exponential-delay` with 10 attempts, which retries for about two
minutes before the statement fails. Operators can change this default, as
described in [Configure the default restart strategy](#flink-sql-batch-restart-strategy).

When Flink restarts the tasks of a batch job, the restarted tasks write their
output again. A sink that is not transactional, such as an at-least-once Kafka
sink, can then receive duplicate rows.

A batch `SELECT` statement gets no default restart strategy. Unless you
configure one, its job fails on the first task failure, such as a lost
TaskManager.

<a id="flink-sql-batch-known-limitations"></a>

## Known limitations

The following problems are known in CMF 2.5. Each has a workaround.

Bounding a source in a window function
: Flink does not accept a table hint on the `TABLE` argument of a window
  function such as `TUMBLE`. The statement fails with `SQL parse failed.
  Encountered "/*+"`. Instead, create a view in the
  [environment catalog](../../configure/environment-catalog.md#cmf-environment-catalog) that reads the table
  with the hint, and run the window function over the view:
  <br/>
  ```sql
  CREATE VIEW orders_bounded AS
    SELECT * FROM `kafka-cat`.`kafka-db`.`orders`
    /*+ OPTIONS('scan.startup.mode'='earliest-offset',
               'scan.bounded.mode'='latest-offset') */;
  <br/>
  SELECT window_start, window_end, COUNT(*)
    FROM TABLE(
      TUMBLE(TABLE orders_bounded, DESCRIPTOR(ts), INTERVAL '10' SECOND))
    GROUP BY window_start, window_end;
  ```
  <br/>
  `SELECT *` does not include the `$rowtime` column of a Kafka table. If
  the window uses `$rowtime`, add it to the view’s select list.
  <br/>
  You can also set `scan.bounded.mode` on the table itself with
  `ALTER TABLE`, but then every statement that reads the table, including
  streaming statements, reads it as a bounded source.

Bounded-mode options are not checked
: CMF does not check the `scan.bounded.*` options. A missing or wrong
  option does not fail the statement at create:
  <br/>
  * `'scan.bounded.mode'='timestamp'` without
    `scan.bounded.timestamp-millis` reads no rows, and the statement
    completes with an empty result.
  * `'scan.bounded.mode'='specific-offsets'` with an offset beyond the end of
    a partition never finishes. Without `scan.bounded.specific-offsets`, the
    job fails after it starts.
  * `'scan.bounded.mode'='group-offsets'` without `properties.group.id`
    fails after the job starts.
  * A misspelled option name fails with the error that the query requires
    bounded sources, and does not name the option.
  <br/>
  Use `latest-offset` unless you need another end point, and set the option
  that each mode requires, as described in [Table options](table-operations.md#table-options).

Parallelism below the compute pool’s minimum
: If the compute pool sets
  `execution.batch.adaptive.auto-parallelism.min-parallelism` higher than the
  statement’s `parallelism`, which defaults to 1, the job fails at start with
  `Maximum parallelism should be greater than or equal to the minimum
  parallelism`. Set the statement’s `parallelism` to at least the pool’s
  minimum.

Joins followed by `GROUP BY` on the Default scheduler
: On a compute pool that sets `jobmanager.scheduler: default`, a batch join
  followed by a `GROUP BY` on the join key fails at job start with
  `Could not start the JobMaster`. Do not set `jobmanager.scheduler` on
  compute pools that run batch statements, so that Flink uses its default
  adaptive batch scheduler.

Runtime filters
: With `table.optimizer.runtime-filter.enabled: true`, a batch join fails to
  compile, and the statement fails with `Unable to serialize given
  ExecNodeGraph`. Do not set this option.

The `NEST_LOOP` join hint
: A batch query with a `/*+ NEST_LOOP(...) */` hint compiles, but its job
  fails. Leave the join strategy to the planner, or use the `SHUFFLE_HASH` or
  `SHUFFLE_MERGE` hint.

Batch job recovery without high availability
: If a compute pool sets `execution.batch.job-recovery.enabled: true` but no
  `high-availability.storageDir`, its JobManager fails at start, and every
  statement on the pool fails. The error appears only in the JobManager log.
  Set `high-availability.storageDir` on any pool that enables batch job
  recovery.

Reading Iceberg tables
: A batch statement that reads an Iceberg table can start many TaskManagers
  for a small query, and can fail with `Insufficient number of network
  buffers`. Add `table.exec.iceberg.infer-source-parallelism: false` to the
  statement’s `flinkConfiguration`.

<a id="flink-sql-batch-restart-strategy"></a>

## Configure the default restart strategy

Operators set the restart strategy that CMF adds to batch `INSERT INTO`
statements and statement sets with the `cmf.sql.batchRestartStrategy` Helm
value.

| Helm value                                        | Default                              | Description                                                                                                                                                                                                                                                                                          |
|---------------------------------------------------|--------------------------------------|------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
| `cmf.sql.batchRestartStrategy.enabled`            | `true`                               | Set to `false` to add no restart strategy.                                                                                                                                                                                                                                                           |
| `cmf.sql.batchRestartStrategy.flinkConfiguration` | `exponential-delay` with 10 attempts | The Flink restart strategy options to add. Only `restart-strategy.*`<br/>options are allowed, and `restart-strategy.type` is required. The<br/>default sets `restart-strategy.type` to `exponential-delay` and<br/>`restart-strategy.exponential-delay.attempts-before-reset-backoff`<br/>to `"10"`. |

CMF adds these options only when none of the following sets
`restart-strategy.type`: the statement, the environment’s statement
defaults, the compute pool, and the environment’s compute pool defaults. A
restart strategy that you set in any of these places is kept. CMF does not
store the options in the statement, so they do not appear in the statement’s
`flinkConfiguration`. They appear in the job configuration in the Flink web
UI.

To use a fixed delay between restarts instead, follow these steps:

1. Add the following to your Helm values file. Quote numbers, because the
   values must be strings.
   ```yaml
   cmf:
     sql:
       batchRestartStrategy:
         enabled: true
         flinkConfiguration:
           restart-strategy.type: fixed-delay
           restart-strategy.fixed-delay.attempts: "5"
           restart-strategy.fixed-delay.delay: "30 s"
   ```
2. Upgrade CMF:
   ```bash
   helm upgrade --install cmf confluentinc/confluent-manager-for-apache-flink \
     --version "~2.5.0" \
     --namespace <namespace> \
     -f values.yaml
   ```

Helm merges your `flinkConfiguration` with the default, so the default’s
`restart-strategy.exponential-delay.attempts-before-reset-backoff` option is
still added. Flink ignores it when `restart-strategy.type` is not
`exponential-delay`.

The chart’s schema rejects most invalid values when you run `helm upgrade`:
options that are not Flink restart strategy options, values that are not
strings, and a missing `restart-strategy.type`. If a value passes the schema
but Flink rejects it, CMF does not start, and its log names the option.

The change applies to batch statements that start after the upgrade. A batch
job that is already running finishes with the restart strategy it started
with.

## Related content

* [Create Statements with Confluent Manager for Apache Flink](create-statements.md#flink-sql-statements-create)
* [Manage Statements in Confluent Manager for Apache Flink](manage-statements.md#flink-sql-statements-manage)
* [Table options](table-operations.md#table-options)
* [Configure Environments in Confluent Manager for Apache Flink](../../configure/environments.md#cmf-environments)
* [Savepoint Management in Confluent Manager for Apache Flink](../savepoints.md#cmf-savepoints)