private void processErrorPosition(SQLException e) throws SQLException {
        List<Integer> errorPositions = parseError(e);
        // the data before the first sqlexception are already written, handle those and remove them.
        int writtenSize = errorPositions.get(0);
        long writtenBytes = 0L;
        if (writtenSize > 0) {
            writtenBytes = CalculateObjectSizeUtils.getDataSize(batch.get(0)) * writtenSize;
        }
        if (!multipleSink) {
            sinkMetricData.invoke(writtenSize, writtenBytes);
        } else {
            metric[0] += writtenSize;
            metric[1] += writtenBytes;
        }

        batch = batch.subList(writtenSize, batch.size());

        // for the unwritten data, remove the dirty ones
        for (int pos : errorPositions) {
            pos -= writtenSize;
            RowData record = batch.get(pos);
            batch.remove(record);
            invokeDirty(record, e);
        }

        // try to execute the supposedly clean batch, throw exception on failure
        for (RowData record : batch) {
            addToBatch(record);
        }
        st.executeBatch();
        batch.clear();
        st.clearParameters();
    }
A {@link JdbcBatchStatementExecutor} that simply adds the records into batches of {@link
java.sql.PreparedStatement} and doesn't buffer records in memory. Only used in Table/SQL API.
Supported executors:TableBufferedStatementExecutor, TableBufferReducedStatementExecutor, TableSimpleStatementExecutor
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/jdbc/src/main/java/org/apache/inlong/sort/jdbc/internal/TableMetricStatementExecutor.java