<commit_before>/**
 * Copyright (C) 2011 Akiban Technologies Inc.
 * This program is free software: you can redistribute it and/or modify
 * it under the terms of the GNU Affero General Public License, version 3,
 * as published by the Free Software Foundation.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU Affero General Public License for more details.
 *
 * You should have received a copy of the GNU Affero General Public License
 * along with this program.  If not, see http://www.gnu.org/licenses.
 */

package com.akiban.server.mttests.mtddl;

import com.akiban.ais.model.Index;
import com.akiban.ais.model.TableName;
import com.akiban.server.InvalidOperationException;
import com.akiban.server.api.dml.EasyUseColumnSelector;
import com.akiban.server.api.dml.NoSuchIndexException;
import com.akiban.server.api.dml.scan.CursorId;
import com.akiban.server.api.dml.scan.NewRow;
import com.akiban.server.api.dml.scan.NiceRow;
import com.akiban.server.api.dml.scan.RowOutput;
import com.akiban.server.api.dml.scan.RowOutputException;
import com.akiban.server.api.dml.scan.ScanAllRequest;
import com.akiban.server.api.dml.scan.ScanFlag;
import com.akiban.server.itests.ApiTestBase;
import com.akiban.server.mttests.mtutil.TimePoints;
import com.akiban.server.mttests.mtutil.TimePointsComparison;
import com.akiban.server.mttests.mtutil.TimedCallable;
import com.akiban.server.mttests.mtutil.Timing;
import com.akiban.server.mttests.mtutil.TimedResult;
import com.akiban.server.service.session.Session;
import com.akiban.server.service.session.SessionImpl;
import org.junit.Test;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.EnumSet;
import java.util.HashSet;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

import static org.junit.Assert.assertEquals;

public final class ConcurrencyAtomicsMT extends ApiTestBase {

    private static final String SCHEMA = "cold";
    private static final String TABLE = "frosty";

    @Test
    public void dropIndexWhileScanning() throws Exception {
        final int tableId = tableWithTwoRows();
        final int SCAN_WAIT = 10000;

        int indexId = ddl().getUserTable(session(), new TableName(SCHEMA, TABLE)).getIndex("name").getIndexId();
        TimedCallable<List<NewRow>> scanCallable = getScanCallable(tableId, indexId, SCAN_WAIT);
        TimedCallable<Void> dropIndexCallable = new TimedCallable<Void>() {
            @Override
            protected Void doCall(TimePoints timePoints) throws Exception {
                TableName table = new TableName(SCHEMA, TABLE);
                Timing.sleep(2000);
                timePoints.mark("INDEX: DROP>");
                ddl().dropIndexes(new SessionImpl(), table, Collections.singleton("name"));
                timePoints.mark("INDEX: <DROP");
                return null;
            }
        };

        ExecutorService executor = Executors.newFixedThreadPool(2);
        Future<TimedResult<List<NewRow>>> scanFuture = executor.submit(scanCallable);
        Future<TimedResult<Void>> dropIndexFuture = executor.submit(dropIndexCallable);

        TimedResult<List<NewRow>> scanResult = scanFuture.get();
        TimedResult<Void> dropIndexResult = dropIndexFuture.get();

        new TimePointsComparison(scanResult, dropIndexResult).verify(
                "SCAN: START",
                "SCAN: PAUSE",
                "INDEX: DROP>",
                "INDEX: <DROP",
                "SCAN: FINISH"
        );

        List<NewRow> rowsScanned = scanResult.getItem();
        List<NewRow> rowsExpected = Arrays.asList(
                createNewRow(tableId, 2L, "<NAME>"),
                createNewRow(tableId, 1L, "<NAME>")
        );
        assertEquals("rows scanned (in order)", rowsExpected, rowsScanned);
    }

    @Test(timeout=60000)
    public void scanWhileDroppingIndex() throws Exception {
        final int NUMBER_OF_ROWS = 10000;
        final int initialTableId = createTable(SCHEMA, TABLE, "id int key", "age int", "key(age)");
        final TableName tableName = new TableName(SCHEMA, TABLE);
        for(int i=0; i < NUMBER_OF_ROWS; ++i) {
            writeRows(createNewRow(initialTableId, i, i + 1));
        }

        while(true) {
            final Index index = ddl().getUserTable(session(), tableName).getIndex("age");
            final Collection<String> indexNameCollection = Collections.singleton(index.getIndexName().getName());
            final int tableId = ddl().getTableId(session(), tableName);
            assertEquals("table id changed", initialTableId, tableId);
            
            TimedCallable<Void> dropIndexCallable = new TimedCallable<Void>() {
                @Override
                protected Void doCall(TimePoints timePoints) throws Exception {
                    timePoints.mark("DROP: IN");
                    ddl().dropIndexes(new SessionImpl(), tableName, indexNameCollection);
                    timePoints.mark("DROP: OUT");
                    return null;
                }
            };
            TimedCallable<List<NewRow>> scanCallable = getScanCallable(tableId, index.getIndexId(), 0);

            ExecutorService executor = Executors.newFixedThreadPool(2);
            Future<TimedResult<List<NewRow>>> scanFuture = executor.submit(scanCallable);
            Future<TimedResult<Void>> dropIndexFuture = executor.submit(dropIndexCallable);

            TimedResult<List<NewRow>> scanResult = scanFuture.get();
            TimedResult<Void> dropIndexResult = dropIndexFuture.get();

            TimePointsComparison comparison = new TimePointsComparison(scanResult, dropIndexResult);
            if (comparison.matches(
                    "DROP: IN",
                    "SCAN: NO SUCH INDEX",
                    "DROP: OUT"
            )) {
                return; // this is what we wanted to find
            }
            else {
                assertEquals("number of rows scanned", NUMBER_OF_ROWS, scanResult.getItem().size());
            }
            ddl().createIndexes(session(), Collections.singleton(index));
        }
    }

    @Test
    public void updatePKColumnWhileScanning() throws Exception {
        final int tableId = tableWithTwoRows();
        final int SCAN_WAIT = 5000;

        int indexId = ddl().getUserTable(session(), new TableName(SCHEMA, TABLE)).getIndex("PRIMARY").getIndexId();
        TimedCallable<List<NewRow>> scanCallable = getScanCallable(tableId, indexId, SCAN_WAIT);
        TimedCallable<Void> updateCallable = new TimedCallable<Void>() {
            @Override
            protected Void doCall(TimePoints timePoints) throws Exception {
                NewRow old = new NiceRow(tableId);
                old.put(0, 1L);
                NewRow updated = new NiceRow(tableId);
                updated.put(0, 5L);
                Timing.sleep(2000);
                timePoints.mark("UPDATE: IN");
                dml().updateRow(new SessionImpl(), old, updated, new EasyUseColumnSelector(0));
                timePoints.mark("UPDATE: OUT");
                return null;
            }
        };

        ExecutorService executor = Executors.newFixedThreadPool(2);
        Future<TimedResult<List<NewRow>>> scanFuture = executor.submit(scanCallable);
        Future<TimedResult<Void>> updateFuture = executor.submit(updateCallable);

        TimedResult<List<NewRow>> scanResult = scanFuture.get();
        TimedResult<Void> updateResult = updateFuture.get();

        new TimePointsComparison(scanResult, updateResult).verify(
                "SCAN: START",
                "SCAN: PAUSE",
                "UPDATE: IN",
                "UPDATE: OUT",
                "SCAN: FINISH"
        );

        List<NewRow> rowsScanned = scanResult.getItem();
        List<NewRow> rowsExpected = Arrays.asList(
                createNewRow(tableId, 1L, "the snowman"),
                createNewRow(tableId, 2L, "<NAME>"),
                createNewRow(tableId, 5L, "the snowman")
        );
        assertEquals("rows scanned (in order)", rowsExpected, rowsScanned);

        expectFullRows(tableId,
                createNewRow(tableId, 2L, "<NAME>"),
                createNewRow(tableId, 5L, "the snowman")
        );
    }

    @Test
    public void updateIndexedColumnWhileScanning() throws Exception {
        final int tableId = tableWithTwoRows();
        final int SCAN_WAIT = 5000;

        int indexId = ddl().getUserTable(session(), new TableName(SCHEMA, TABLE)).getIndex("name").getIndexId();
        TimedCallable<List<NewRow>> scanCallable = getScanCallable(tableId, indexId, SCAN_WAIT);
        TimedCallable<Void> updateCallable = new TimedCallable<Void>() {
            @Override
            protected Void doCall(TimePoints timePoints) throws Exception {
                NewRow old = new NiceRow(tableId);
                old.put(0, 2L);
                NewRow updated = new NiceRow(tableId);
                updated.put(0, 2L);
                updated.put(1, "xtreme weather");
                Timing.sleep(2000);
                timePoints.mark("UPDATE: IN");
                dml().updateRow(new SessionImpl(), old, updated, new EasyUseColumnSelector(1));
                timePoints.mark("UPDATE: OUT");
                return null;
            }
        };

        ExecutorService executor = Executors.newFixedThreadPool(2);
        Future<TimedResult<List<NewRow>>> scanFuture = executor.submit(scanCallable);
        Future<TimedResult<Void>> updateFuture = executor.submit(updateCallable);

        TimedResult<List<NewRow>> scanResult = scanFuture.get();
        TimedResult<Void> updateResult = updateFuture.get();

        new TimePointsComparison(scanResult, updateResult).verify(
                "SCAN: START",
                "SCAN: PAUSE",
                "UPDATE: IN",
                "UPDATE: OUT",
                "SCAN: FINISH"
        );

        List<NewRow> rowsScanned = scanResult.getItem();
        List<NewRow> rowsExpected = Arrays.asList(
                createNewRow(tableId, 2L, "<NAME>"),
                createNewRow(tableId, 1L, "the snowman"),
                createNewRow(tableId, 2L, "xtreme weather")
        );
        assertEquals("rows scanned (in order)", rowsExpected, rowsScanned);

        expectFullRows(tableId,
                createNewRow(tableId, 1L, "the snowman"),
                createNewRow(tableId, 2L, "xtreme weather")
        );
    }

    @Test
    public void updateUnIndexedColumnWhileScanning() throws Exception {
        final int tableId = tableWithTwoRows();
        final int SCAN_WAIT = 5000;

        int indexId = ddl().getUserTable(session(), new TableName(SCHEMA, TABLE)).getIndex("PRIMARY").getIndexId();
        TimedCallable<List<NewRow>> scanCallable = getScanCallable(tableId, indexId, SCAN_WAIT);
        TimedCallable<Void> updateCallable = new TimedCallable<Void>() {
            @Override
            protected Void doCall(TimePoints timePoints) throws Exception {
                NewRow old = new NiceRow(tableId);
                old.put(0, 2L);
                NewRow updated = new NiceRow(tableId);
                updated.put(0, 2L);
                updated.put(1, "icebox");
                Timing.sleep(2000);
                timePoints.mark("UPDATE: IN");
                dml().updateRow(new SessionImpl(), old, updated, new EasyUseColumnSelector(1));
                timePoints.mark("UPDATE: OUT");
                return null;
            }
        };

        ExecutorService executor = Executors.newFixedThreadPool(2);
        Future<TimedResult<List<NewRow>>> scanFuture = executor.submit(scanCallable);
        Future<TimedResult<Void>> updateFuture = executor.submit(updateCallable);

        TimedResult<List<NewRow>> scanResult = scanFuture.get();
        TimedResult<Void> updateResult = updateFuture.get();

        new TimePointsComparison(scanResult, updateResult).verify(
                "SCAN: START",
                "SCAN: PAUSE",
                "UPDATE: IN",
                "UPDATE: OUT",
                "SCAN: FINISH"
        );

        List<NewRow> rowsScanned = scanResult.getItem();
        List<NewRow> rowsExpected = Arrays.asList(
                createNewRow(tableId, 1L, "the snowman"),
                createNewRow(tableId, 2L, "icebox")
        );
        assertEquals("rows scanned (in order)", rowsExpected, rowsScanned);

        expectFullRows(tableId,
                createNewRow(tableId, 1L, "the snowman"),
                createNewRow(tableId, 2L, "icebox")
        );

    }

    private TimedCallable<List<NewRow>> getScanCallable(final int tableId, final int indexId, final int sleepBetween) {
        return new TimedCallable<List<NewRow>>() {
            @Override
            protected List<NewRow> doCall(TimePoints timePoints) throws Exception {
                Session session = new SessionImpl();
                ScanAllRequest request = new ScanAllRequest(
                        tableId,
                        new HashSet<Integer>(Arrays.asList(0, 1)),
                        indexId,
                        EnumSet.of(ScanFlag.START_AT_BEGINNING, ScanFlag.END_AT_END)
                );
                final CursorId cursorId;
                try {
                     cursorId = dml().openCursor(session, request);
                } catch (NoSuchIndexException e) {
                    timePoints.mark("SCAN: NO SUCH INDEX");
                    return Collections.emptyList();
                }
                CountingRowOutput output = new CountingRowOutput();
                timePoints.mark("SCAN: START");
                if (!dml().scanSome(session, cursorId, output, 1)) {
                    timePoints.mark("SCAN: EARLY FINISH");
                    return output.rows;
                }
                if (sleepBetween > 0) {
                    timePoints.mark("SCAN: PAUSE");
                    Timing.sleep(sleepBetween);
                }
                dml().scanSome(session, cursorId, output, -1);
                timePoints.mark("SCAN: FINISH");

                return output.rows;
            }
        };
    }

    int tableWithTwoRows() throws InvalidOperationException {
        int id = createTable(SCHEMA, TABLE, "id int key", "name varchar(32)", "key(name)");
        writeRows(
            createNewRow(id, 1L, "the snowman"),
            createNewRow(id, 2L, "<NAME>")
        );
        return id;
    }


    private static class CountingRowOutput implements RowOutput {
        private final List<NewRow> rows = new ArrayList<NewRow>();
        @Override
        public void output(NewRow row) throws RowOutputException {
           rows.add(row);
        }
    }
}
<commit_msg>Refactoring.<commit_after>/**
 * Copyright (C) 2011 Akiban Technologies Inc.
 * This program is free software: you can redistribute it and/or modify
 * it under the terms of the GNU Affero General Public License, version 3,
 * as published by the Free Software Foundation.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU Affero General Public License for more details.
 *
 * You should have received a copy of the GNU Affero General Public License
 * along with this program.  If not, see http://www.gnu.org/licenses.
 */

package com.akiban.server.mttests.mtddl;

import com.akiban.ais.model.Index;
import com.akiban.ais.model.TableName;
import com.akiban.server.InvalidOperationException;
import com.akiban.server.api.dml.EasyUseColumnSelector;
import com.akiban.server.api.dml.NoSuchIndexException;
import com.akiban.server.api.dml.scan.CursorId;
import com.akiban.server.api.dml.scan.NewRow;
import com.akiban.server.api.dml.scan.NiceRow;
import com.akiban.server.api.dml.scan.RowOutput;
import com.akiban.server.api.dml.scan.RowOutputException;
import com.akiban.server.api.dml.scan.ScanAllRequest;
import com.akiban.server.api.dml.scan.ScanFlag;
import com.akiban.server.itests.ApiTestBase;
import com.akiban.server.mttests.mtutil.TimePoints;
import com.akiban.server.mttests.mtutil.TimePointsComparison;
import com.akiban.server.mttests.mtutil.TimedCallable;
import com.akiban.server.mttests.mtutil.Timing;
import com.akiban.server.mttests.mtutil.TimedResult;
import com.akiban.server.service.session.Session;
import com.akiban.server.service.session.SessionImpl;
import org.junit.Test;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.EnumSet;
import java.util.HashSet;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

import static org.junit.Assert.assertEquals;

public final class ConcurrencyAtomicsMT extends ApiTestBase {

    private static final String SCHEMA = "cold";
    private static final String TABLE = "frosty";

    @Test
    public void dropIndexWhileScanning() throws Exception {
        final int tableId = tableWithTwoRows();
        final int SCAN_WAIT = 10000;

        int indexId = ddl().getUserTable(session(), new TableName(SCHEMA, TABLE)).getIndex("name").getIndexId();
        TimedCallable<List<NewRow>> scanCallable = getScanCallable(tableId, indexId, SCAN_WAIT);
        TimedCallable<Void> dropIndexCallable = new TimedCallable<Void>() {
            @Override
            protected Void doCall(TimePoints timePoints) throws Exception {
                TableName table = new TableName(SCHEMA, TABLE);
                Timing.sleep(2000);
                timePoints.mark("INDEX: DROP>");
                ddl().dropIndexes(new SessionImpl(), table, Collections.singleton("name"));
                timePoints.mark("INDEX: <DROP");
                return null;
            }
        };

        ExecutorService executor = Executors.newFixedThreadPool(2);
        Future<TimedResult<List<NewRow>>> scanFuture = executor.submit(scanCallable);
        Future<TimedResult<Void>> dropIndexFuture = executor.submit(dropIndexCallable);

        TimedResult<List<NewRow>> scanResult = scanFuture.get();
        TimedResult<Void> dropIndexResult = dropIndexFuture.get();

        new TimePointsComparison(scanResult, dropIndexResult).verify(
                "SCAN: START",
                "SCAN: PAUSE",
                "INDEX: DROP>",
                "INDEX: <DROP",
                "SCAN: FINISH"
        );

        List<NewRow> rowsScanned = scanResult.getItem();
        List<NewRow> rowsExpected = Arrays.asList(
                createNewRow(tableId, 2L, "<NAME>"),
                createNewRow(tableId, 1L, "<NAME>")
        );
        assertEquals("rows scanned (in order)", rowsExpected, rowsScanned);
    }

    @Test(timeout=60000)
    public void scanWhileDroppingIndex() throws Exception {
        final int NUMBER_OF_ROWS = 10000;
        final int initialTableId = createTable(SCHEMA, TABLE, "id int key", "age int", "key(age)");
        final TableName tableName = new TableName(SCHEMA, TABLE);
        for(int i=0; i < NUMBER_OF_ROWS; ++i) {
            writeRows(createNewRow(initialTableId, i, i + 1));
        }

        while(true) {
            final Index index = ddl().getUserTable(session(), tableName).getIndex("age");
            final Collection<String> indexNameCollection = Collections.singleton(index.getIndexName().getName());
            final int tableId = ddl().getTableId(session(), tableName);
            assertEquals("table id changed", initialTableId, tableId);
            
            TimedCallable<Void> dropIndexCallable = new TimedCallable<Void>() {
                @Override
                protected Void doCall(TimePoints timePoints) throws Exception {
                    timePoints.mark("DROP: IN");
                    ddl().dropIndexes(new SessionImpl(), tableName, indexNameCollection);
                    timePoints.mark("DROP: OUT");
                    return null;
                }
            };
            TimedCallable<List<NewRow>> scanCallable = getScanCallable(tableId, index.getIndexId(), 0);

            ExecutorService executor = Executors.newFixedThreadPool(2);
            Future<TimedResult<List<NewRow>>> scanFuture = executor.submit(scanCallable);
            Future<TimedResult<Void>> dropIndexFuture = executor.submit(dropIndexCallable);

            TimedResult<List<NewRow>> scanResult = scanFuture.get();
            TimedResult<Void> dropIndexResult = dropIndexFuture.get();

            TimePointsComparison comparison = new TimePointsComparison(scanResult, dropIndexResult);
            if (comparison.matches(
                    "DROP: IN",
                    "SCAN: NO SUCH INDEX",
                    "DROP: OUT"
            )) {
                return; // this is what we wanted to find
            }
            else {
                assertEquals("number of rows scanned", NUMBER_OF_ROWS, scanResult.getItem().size());
            }
            ddl().createIndexes(session(), Collections.singleton(index));
        }
    }

    @Test
    public void updatePKColumnWhileScanning() throws Exception {
        final int tableId = tableWithTwoRows();
        final int SCAN_WAIT = 5000;

        int indexId = ddl().getUserTable(session(), new TableName(SCHEMA, TABLE)).getIndex("PRIMARY").getIndexId();
        TimedCallable<List<NewRow>> scanCallable = getScanCallable(tableId, indexId, SCAN_WAIT);
        TimedCallable<Void> updateCallable = new TimedCallable<Void>() {
            @Override
            protected Void doCall(TimePoints timePoints) throws Exception {
                NewRow old = new NiceRow(tableId);
                old.put(0, 1L);
                NewRow updated = new NiceRow(tableId);
                updated.put(0, 5L);
                Timing.sleep(2000);
                timePoints.mark("UPDATE: IN");
                dml().updateRow(new SessionImpl(), old, updated, new EasyUseColumnSelector(0));
                timePoints.mark("UPDATE: OUT");
                return null;
            }
        };

        scanUpdateConfirm(
                tableId,
                scanCallable,
                updateCallable,
                Arrays.asList(
                        createNewRow(tableId, 1L, "the snowman"),
                        createNewRow(tableId, 2L, "<NAME>"),
                        createNewRow(tableId, 5L, "the snowman")
                ),
                Arrays.asList(
                        createNewRow(tableId, 2L, "<NAME>"),
                        createNewRow(tableId, 5L, "the snowman")
                )
        );
    }

    @Test
    public void updateIndexedColumnWhileScanning() throws Exception {
        final int tableId = tableWithTwoRows();
        final int SCAN_WAIT = 5000;

        int indexId = ddl().getUserTable(session(), new TableName(SCHEMA, TABLE)).getIndex("name").getIndexId();
        TimedCallable<List<NewRow>> scanCallable = getScanCallable(tableId, indexId, SCAN_WAIT);
        TimedCallable<Void> updateCallable = new TimedCallable<Void>() {
            @Override
            protected Void doCall(TimePoints timePoints) throws Exception {
                NewRow old = new NiceRow(tableId);
                old.put(0, 2L);
                NewRow updated = new NiceRow(tableId);
                updated.put(0, 2L);
                updated.put(1, "xtreme weather");
                Timing.sleep(2000);
                timePoints.mark("UPDATE: IN");
                dml().updateRow(new SessionImpl(), old, updated, new EasyUseColumnSelector(1));
                timePoints.mark("UPDATE: OUT");
                return null;
            }
        };

        scanUpdateConfirm(
                tableId,
                scanCallable,
                updateCallable,
                Arrays.asList(
                        createNewRow(tableId, 2L, "mr melty"),
                        createNewRow(tableId, 1L, "the snowman"),
                        createNewRow(tableId, 2L, "xtreme weather")
                ),
                Arrays.asList(
                        createNewRow(tableId, 1L, "the snowman"),
                        createNewRow(tableId, 2L, "xtreme weather")
                )
        );
    }

    @Test
    public void updateUnIndexedColumnWhileScanning() throws Exception {
        final int tableId = tableWithTwoRows();
        final int SCAN_WAIT = 5000;

        int indexId = ddl().getUserTable(session(), new TableName(SCHEMA, TABLE)).getIndex("PRIMARY").getIndexId();
        TimedCallable<List<NewRow>> scanCallable = getScanCallable(tableId, indexId, SCAN_WAIT);
        TimedCallable<Void> updateCallable = new TimedCallable<Void>() {
            @Override
            protected Void doCall(TimePoints timePoints) throws Exception {
                NewRow old = new NiceRow(tableId);
                old.put(0, 2L);
                NewRow updated = new NiceRow(tableId);
                updated.put(0, 2L);
                updated.put(1, "icebox");
                Timing.sleep(2000);
                timePoints.mark("UPDATE: IN");
                dml().updateRow(new SessionImpl(), old, updated, new EasyUseColumnSelector(1));
                timePoints.mark("UPDATE: OUT");
                return null;
            }
        };

        scanUpdateConfirm(
                tableId,
                scanCallable,
                updateCallable,
                Arrays.asList(
                        createNewRow(tableId, 1L, "the snowman"),
                        createNewRow(tableId, 2L, "icebox")
                ),
                Arrays.asList(
                        createNewRow(tableId, 1L, "the snowman"),
                        createNewRow(tableId, 2L, "icebox")
                )
        );
    }

    private void scanUpdateConfirm(int tableId,
                                   TimedCallable<List<NewRow>> scanCallable,
                                   TimedCallable<Void> updateCallable,
                                   List<NewRow> scanCallableExpected,
                                   List<NewRow> endStateExpected)
            throws Exception
    {
        ExecutorService executor = Executors.newFixedThreadPool(2);
        Future<TimedResult<List<NewRow>>> scanFuture = executor.submit(scanCallable);
        Future<TimedResult<Void>> updateFuture = executor.submit(updateCallable);

        TimedResult<List<NewRow>> scanResult = scanFuture.get();
        TimedResult<Void> updateResult = updateFuture.get();

        new TimePointsComparison(scanResult, updateResult).verify(
                "SCAN: START",
                "SCAN: PAUSE",
                "UPDATE: IN",
                "UPDATE: OUT",
                "SCAN: FINISH"
        );

        List<NewRow> rowsScanned = scanResult.getItem();
        assertEquals("rows scanned (in order)", scanCallableExpected, rowsScanned);

        expectFullRows(tableId, endStateExpected.toArray( new NewRow[endStateExpected.size()] ));
    }

    private TimedCallable<List<NewRow>> getScanCallable(final int tableId, final int indexId, final int sleepBetween) {
        return new TimedCallable<List<NewRow>>() {
            @Override
            protected List<NewRow> doCall(TimePoints timePoints) throws Exception {
                Session session = new SessionImpl();
                ScanAllRequest request = new ScanAllRequest(
                        tableId,
                        new HashSet<Integer>(Arrays.asList(0, 1)),
                        indexId,
                        EnumSet.of(ScanFlag.START_AT_BEGINNING, ScanFlag.END_AT_END)
                );
                final CursorId cursorId;
                try {
                     cursorId = dml().openCursor(session, request);
                } catch (NoSuchIndexException e) {
                    timePoints.mark("SCAN: NO SUCH INDEX");
                    return Collections.emptyList();
                }
                CountingRowOutput output = new CountingRowOutput();
                timePoints.mark("SCAN: START");
                if (!dml().scanSome(session, cursorId, output, 1)) {
                    timePoints.mark("SCAN: EARLY FINISH");
                    return output.rows;
                }
                if (sleepBetween > 0) {
                    timePoints.mark("SCAN: PAUSE");
                    Timing.sleep(sleepBetween);
                }
                dml().scanSome(session, cursorId, output, -1);
                timePoints.mark("SCAN: FINISH");

                return output.rows;
            }
        };
    }

    int tableWithTwoRows() throws InvalidOperationException {
        int id = createTable(SCHEMA, TABLE, "id int key", "name varchar(32)", "key(name)");
        writeRows(
            createNewRow(id, 1L, "the snowman"),
            createNewRow(id, 2L, "<NAME>")
        );
        return id;
    }


    private static class CountingRowOutput implements RowOutput {
        private final List<NewRow> rows = new ArrayList<NewRow>();
        @Override
        public void output(NewRow row) throws RowOutputException {
           rows.add(row);
        }
    }
}

{'fraction_non_alphanumeric': 0.10381694984507474, 'fraction_numerical': 0.0066056045490142665, 'mean_word_length': 2.181670458238544, 'pattern_counts': {'":': 0, '<': 158, '<?xml version=': 0, '>': 158, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1892072', 'n_tokens_mistral': 8340, 'n_tokens_neox': 7852, 'n_words': 1747}