# include "dvosplit.h"
# define NROWS 100000 /* ~10MB per row for measures */
# define DNOUT 1000

int split_measures (Catalog *incatalog, SkyList *outlist, Catalog *outcatalogs, AveLinks *avelinks) {

  int block, meas, cat, Nblocks, Ncat, Nout, averef;
  int *outref, *outcat, *outmem;

  outref = avelinks->outref;
  outcat = avelinks->outcat;
  ALLOCATE (outmem, int, outlist[0].Nregions);

  // allocate enough space for the output buffer
  for (cat = 0; cat < outlist[0].Nregions; cat++) {
    outmem[cat] = DNOUT;
    REALLOCATE (outcatalogs[cat].measure, Measure, outmem[cat]);
  }

  // split out the measure entries:
  incatalog[0].catflags = LOAD_MEAS;
  Nblocks = incatalog[0].Nmeas_disk / NROWS;
  if (incatalog[0].Nmeas_disk % NROWS) Nblocks ++;
  for (block = 0; block < Nblocks; block++) {

    // read up to NROWS at a time
    dvo_catalog_load_segment (incatalog, VERBOSE, block*NROWS, NROWS);
    fprintf (stderr, "splitting %s (measures) .. %d of %d\n", incatalog[0].filename, block, Nblocks);

    assert (block*NROWS == incatalog[0].Nmeas_off);

    for (meas = 0; meas < incatalog[0].Nmeasure; meas++) {

      averef = incatalog[0].measure[meas].averef;
      Ncat = outcat[averef];

      Nout = outcatalogs[Ncat].Nmeasure;
      outcatalogs[Ncat].measure[Nout] = incatalog[0].measure[meas];
      outcatalogs[Ncat].measure[Nout].averef = outref[averef];
      outcatalogs[Ncat].Nmeasure++;

      if (outcatalogs[Ncat].Nmeasure >= outmem[Ncat]) {
	outmem[Ncat] += DNOUT;
	REALLOCATE (outcatalogs[Ncat].measure, Measure, outmem[Ncat]);
      }
    }
    dvo_catalog_free_data (incatalog);

    for (cat = 0; cat < outlist[0].Nregions; cat++) {
      outcatalogs[cat].catflags = LOAD_MEAS;
      dvo_catalog_save (&outcatalogs[cat], VERBOSE);

      outcatalogs[cat].Nmeas_disk += outcatalogs[cat].Nmeasure;
      outcatalogs[cat].Nmeas_off  += outcatalogs[cat].Nmeasure;
      outcatalogs[cat].Nmeasure    = 0;

    }
  }

  for (cat = 0; cat < outlist[0].Nregions; cat++) {
    dvo_catalog_free_data (&outcatalogs[cat]);
  }

  free (outmem);

  return (TRUE);
}
