@Override
    public void executeBatch() throws SQLException {
        try {
            st.executeBatch();

            long writtenSize = batch.size();
            // approximate since it may be inefficient to iterate over all writtenSize-1 elements.
            long writtenBytes = 0L;
            if (writtenSize > 0) {
                writtenBytes = CalculateObjectSizeUtils.getDataSize(batch.get(0)) * writtenSize;
            }
            batch.clear();
            if (!multipleSink) {
                sinkMetricData.invoke(writtenSize, writtenBytes);
            } else {
                metric[0] += writtenSize;
                metric[1] += writtenBytes;
            }

        } catch (SQLException e) {
            // clear the prepared statement first to avoid exceptions
            st.clearParameters();
            try {
                processErrorPosition(e);
            } catch (Exception ex) {
                try {
                    retryEntireBatch();
                } catch (JsonProcessingException exc) {
                    LOG.error("dirty data archive failed");
                }
            }
        }
    }
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