The read-only job that ran out of memory

import java.util.*;

/**
 * Solution: the read-only job that ran out of memory.
 *
 * Two defects, and the second one is only visible because of a check that was
 * already passing. Flushing more often is the instinctive answer to a memory
 * problem and makes it worse — the context still holds everything, and now
 * compares all of it repeatedly.
 */
public class Solution {

    static final class Product {
        final long id;
        String name;
        int priceMinor;
        int stock;
        String category;
        Product(long id, String name, int priceMinor, int stock, String category) {
            this.id = id; this.name = name; this.priceMinor = priceMinor;
            this.stock = stock; this.category = category;
        }
        Map<String, Object> state() {
            var s = new LinkedHashMap<String, Object>();
            s.put("name", name); s.put("priceMinor", priceMinor);
            s.put("stock", stock); s.put("category", category);
            return s;
        }
    }

    /* ── the persistence context (correct; do not change) ───────────────── */

    static final class Session {
        final Map<Long, Product> table = new LinkedHashMap<>();
        final Map<Long, Product> managed = new LinkedHashMap<>();
        final Map<Long, Map<String, Object>> snapshots = new LinkedHashMap<>();

        boolean readOnly, dynamicUpdate;
        int peakSnapshots, comparisons, columnsWritten, statements;

        Session seed(int n) {
            for (long i = 1; i <= n; i++)
                table.put(i, new Product(i, "p" + i, 1000 + (int) i, 5, i % 2 == 0 ? "A" : "B"));
            return this;
        }

        Product find(long id) {
            statements++;
            Product row = table.get(id);
            var entity = new Product(row.id, row.name, row.priceMinor, row.stock, row.category);
            managed.put(id, entity);
            if (!readOnly) snapshots.put(id, entity.state());
            peakSnapshots = Math.max(peakSnapshots, snapshots.size());
            return entity;
        }

        /** Compares every managed entity against its snapshot. */
        void flush() {
            for (var e : managed.entrySet()) {
                var snap = snapshots.get(e.getKey());
                if (snap == null) continue;
                var now = e.getValue().state();
                var changed = new ArrayList<String>();
                for (String f : now.keySet()) {
                    comparisons++;
                    if (!Objects.equals(now.get(f), snap.get(f))) changed.add(f);
                }
                if (changed.isEmpty()) continue;
                statements++;
                columnsWritten += dynamicUpdate ? changed.size() : now.size();
                snapshots.put(e.getKey(), now);
                table.put(e.getKey(), e.getValue());          // the UPDATE lands

            }
        }

        void clear() { managed.clear(); snapshots.clear(); }

        /** A bulk statement: one statement, no entities, no snapshots. */
        void bulkUpdate(String category, int columns, int rows) {
            statements++;
            columnsWritten += columns * rows;
        }

        void commit() { flush(); }
    }

    /* ── the job under review ───────────────────────────────────────────── */

    static final class RepricingJob {
        final Session session;
        RepricingJob(Session session) { this.session = session; }

        static final int BATCH = 500;

        /*
         * Defect 1. Every product loaded stayed in the persistence context,
         * with a snapshot beside it, until the single commit at the end. Two
         * hundred thousand products meant two hundred thousand entities plus
         * two hundred thousand snapshots held simultaneously — which is the
         * OutOfMemoryError, in a job whose ticket said "read-only".
         *
         * flush() alone is not the fix and is the tempting one: it writes the
         * pending changes and keeps every entity managed, so the next flush
         * compares them all again. clear() is what releases them, and check 2
         * exists to catch the version of this fix that omits it.
         */
        void reprice(int productCount) {
            /*
             * Defect 2. The default UPDATE sets every column, so changing one
             * price wrote four columns — which is why the audit table gained a
             * row per product whether or not the price moved. @DynamicUpdate
             * builds the statement from the fields that actually differ.
             */
            session.dynamicUpdate = true;

            for (long id = 1; id <= productCount; id++) {
                Product p = session.find(id);
                if ("A".equals(p.category))
                    p.priceMinor = (int) (p.priceMinor * 1.1);

                if (id % BATCH == 0) {
                    session.flush();
                    session.clear();          // without this, flush makes it worse
                }
            }
            session.commit();
        }
    }

    /* ── checks ─────────────────────────────────────────────────────────── */

    static final int PRODUCTS = 20_000;
    static final int IN_CATEGORY_A = PRODUCTS / 2;

    public static void main(String[] args) {
        List<String> failures = new ArrayList<>();

        var session = new Session().seed(PRODUCTS);
        new RepricingJob(session).reprice(PRODUCTS);

        // 1. The job must not hold the whole table in the persistence context.
        if (session.peakSnapshots > 1_000)
            failures.add("1. the context held " + session.peakSnapshots + " snapshots at once for "
                       + PRODUCTS + " products — nothing is ever released before commit");

        // 2. Flushing more often must not mean comparing the same entities
        //    again. Each entity should be compared once, not once per flush.
        if (session.comparisons > PRODUCTS * 4L)
            failures.add("2. " + session.comparisons + " field comparisons for " + PRODUCTS
                       + " products — entities are being re-compared on every flush, which is what "
                       + "happens when you flush without clearing");

        // 3. Only the price column should be written.
        if (session.columnsWritten > IN_CATEGORY_A)
            failures.add("3. " + session.columnsWritten + " columns were written to change "
                       + IN_CATEGORY_A + " prices — every column is set on every update");

        // 4. And the job must actually do its work.
        long repriced = session.table.values().stream()
            .filter(p -> "A".equals(p.category) && p.priceMinor != 1000 + p.id).count();
        if (repriced != IN_CATEGORY_A)
            failures.add("4. repriced " + repriced + " products, expected " + IN_CATEGORY_A);

        /* ── report ─────────────────────────────────────────────────────── */
        if (failures.isEmpty()) {
            System.out.println("PASS");
        } else {
            failures.forEach(f -> System.out.println("  " + f));
            System.out.println("FAIL");
        }
    }
}