Repository: esheldon/pycolumns (language: python)

Issue:
Newcreate



Discussion / hints:

Fix patch:
diff --git a/README.md b/README.md
index ac35fd0..68f4264 100644
--- a/README.md
+++ b/README.md
@@ -1,11 +1,12 @@
 A simple, efficient column-oriented, pythonic data store.
 
-The focus is currently on efficiency of reading and writing.  The code is pure
-python but searching and reading data is fast due to the use of the fitsio
-package for column data and index data.  Basic consistency is ensured for
-the columns in the table, but the database is not fully ACID.
+The focus is currently on efficiency of reading and writing.  Each table column
+can optionally be compressed using blosc, and can be indexed for fast searches.
+Basic consistency is ensured for the columns in the table, but the database is
+not fully ACID.
 
-The storage is a simple directory with files on disk.
+The storage is a simple directory with files on disk.  The data for each column
+in a table are stored in separate files for simplicity and efficiency.
 
 Examples
 --------
@@ -14,183 +15,348 @@ Examples
 >>> import pycolumns as pyc
 
 # instantiate a column database from the specified coldir
->>> c = pyc.Columns('/some/path/mycols.cols')
+>>> cols = pyc.Columns('/some/path/mycols.cols')
 
 # display some info about the columns
->>> c
-Columns Directory:
-
-  mydata
-  dir: /some/path/mydata.cols
+>>> cols
+Columns:
+  dir: test.cols
   nrows: 64348146
-  Columns:
-    name             dtype index
-    -----------------------------
-    ccd                <i2 True
-    dec                <f8 False
-    exposurename      |S20 True
-    id                 <i8 True
-    imag               <f4 False
-    ra                 <f8 False
-    x                  <f4 False
-    y                  <f4 False
-    g                  <f8 False
-
-  Dictionaries
+
+  Table Columns:
+    name             dtype    comp index
+    -----------------------------------
+    id                 <i8    zstd True
+    name              <U10    zstd True
+    x                  <f4    None False
+    y                  <f4    None False
+
+  Metadata:
     name
-    -----------------------------
+    ----------------------------
     meta
+    versions
 
-  Sub-Columns Directories:
+  Sub Tables:
     name
-    -----------------------------
-    psfstars
+    ----------------------------
+    telemetry/
+
+# Above we see the main types supported:  A table of columns, metadata entries, and
+# sub tables, which are themselves full Columns.  The id and name
+# columns have zstd compression and indexes for fast searching
+#
+# Note there is a main table in the root directory as well as a named
+# table telemetry/ in a subdirectory.  We can print info for it as well
+>>> cols['telemetry/']
+Columns:
+  dir: test.cols/telemetry.cols
+  nrows: 25
+
+  Table Columns:
+    name             dtype    comp index
+    -----------------------------------
+    temp               <f8    None False
+    voltate            <f4    None False
+
+# Only objects in the root are shown above, but we can
+# print the full directory structure which shows that
+# telemetry also has a sub table
+>>> cols.list()
+root has 4 columns 2 metadata
+telemetry/
+  periods/
+
+>>> cols.list(full=True)
+- id
+- name
+- x
+- y
+- {meta}
+- {versions}
+telemetry/
+  - humid
+  - temp
+  periods/
+    - start
+    - end
 
 # display info about column 'id'
->>> c['id']
+>>> cols['id']
 Column:
   name: id
-  filename: ./id.array
+  filename: test.cols/id.array
   type: array
+  index: True
+  compression:
+      cname: zstd
+      clevel: 5
+      shuffle: bitshuffle
+  chunksize: 1m
   dtype: <i8
-  has index: False
   nrows: 64348146
 
 # number of rows in table
->>> c.nrows
+>>> cols.nrows
+64348146
 
-# read all columns into a single rec array.  By default the dict
-# columns are not loaded
+#
+# reading data
+#
 
->>> data = c.read()
+# read all columns into a single structured array.
+>>> data = cols[:]
 
-# using asdict=True puts the data into a dict.  The dict data
-# are also loaded in this case
->>> data = c.read(asdict=True)
+# read a subset of columns into a structured array
+>>> columns = ['id', 'x']
+>>> data = cols[columns][:]
 
-# specify columns
->>> data = c.read(columns=['id', 'flux'])
+# specifying a set of rows as sequence or slice
+>>> rows = [3, 225, 1235]
+>>> data = cols[rows]
+>>> data = cols[20:30]
 
-# dict columns can be specified if asdict is True.  Dicts can also
-# be read as single columns, see below
->>> data = c.read(columns=['id', 'flux', 'meta'], asdict=True)
+# column and row subsets
+>>> data = cols[columns][rows]
+>>> data = cols[columns][10:20]
 
-# specifying a set of rows as sequence/array or slice
->>> data = c.read(columns=['id', 'flux'], rows=[3, 225, 1235])
->>> data = c.read(columns=['id', 'flux'], rows=slice(10, 20))
+# read data from the telemetry/ table
+>>> data = cols['telemetry/'][:]
+>>> v = cols['telemetry/']['voltage'][:]
 
-# read all data from column 'id' as an array rather than recarray
-# alternative syntaxes
->>> ind = c['id'][:]
->>> ind = c['id'].read()
->>> ind = c.read_column('id')
+# You can get the table object directly.  It is also a Columns object
+>>> telcols = cols['telemetry/']
+>>> v = telcols['voltage'][:]
 
-# read a subset of rows
-# slicing
->>> ind = c['id'][25:125]
+# You can also access with a full path
+>>> v = cols['telemetry/voltage'][:]
 
-# specifying a set of rows
->>> rows = [3, 225, 1235]
->>> ind = c['id'][rows]
->>> ind = c.read_column('id', rows=rows)
+#
+# working with single columns
+#
 
-# reading a dictionary column
->>> meta = c['meta'].read()
+# read data from a column as an array rather than as part of a
+# structured array
+>>> x = cols['x'][:]
+>>> x = cols['x'][25:125]
+>>> x = cols['x'][rows]
 
-# Create indexes for fast searching
->>> c['id'].create_index()
+#
+# queries on indexed columns
+#
 
-# get indices for some conditions
->>> ind = c['id'] > 25
->>> ind = c['id'].between(25, 35)
->>> ind = c['id'] == 25
+>>> ind = cols['id'] > 25
+>>> ind = cols['id'].between(25, 35)
+>>> ind = cols['id'] == 25
 
 # read the corresponding data
->>> ccd = c['ccd'][ind]
->>> data = c.read(columns=['ra', 'dec'], rows=ind)
+>>> ccd = cols['ccd'][ind]
+>>> data = cols[columns][ind]
+>>> data = cols.read(columns=['ra', 'dec'], rows=ind)
 
 # composite searches over multiple columns
->>> ind = (c['id'] == 25) & (col['ra'] < 15.23)
->>> ind = c['id'].between(15, 25) | (c['id'] == 55)
->>> ind = c['id'].between(15, 250) & (c['id'] != 66) & (c['ra'] < 100)
+>>> ind = cols['id'].between(15, 25) | (cols['name'] == 'cxj2')
+>>> ind = (cols['id'] != 66) & (cols['name'] != 'af23')
+
+#
+# reading data from a metadata entry.
+#
+
+>>> cols.meta['versions'].read()
+>>> {'data_vers': 'v5b', 's_vers': '1.1.2'}
+
+#
+# As an alternative to slicing you can use the the .read method
+# both for Columns and individual columns
+#
+
+>>> data = cols.read()
+>>> data = cols.read(columns=columns)
+>>> data = cols.read(columns=columns, rows=rows)
+>>> data = cols.read(columns=columns, rows=slice(10, 20))
+>>> x = cols['x'].read()
+>>> x = cols['x'].read(rows=rows)
+
+# using .read you can get the data as a dict of arrays
+>>> data = cols.read(asdict=True)
+
+#
+# Creating a columns data store and adding or updating data
+#
+
+# create an empty Columns
+cols = pyc.Columns.create(coldir)
+
+# Create a new Columns store with a table in the root
+# The schema is determined from input data
+
+dtype = [('id', 'i8'), ('name', 'U10')]
+num = 10
+data = np.zeros(num, dtype=dtype)
+data['id'] = np.arange(num)
+data['name'] = data['id'].astype('U10')
+
+cols = pyc.Columns.create_from_array(coldir, data)
+
+# add more tables not in the root.  from_array() creates the schema from the
+# input array and by default appends the data.  Note table names must end in
+# the / character
+
+# Use default compression
+cols.from_array(data2, name='sub1/', compression=['id', 'name'])
+cols.from_array(data3, name='sub1/sub2/')
+
+# two ways to access sub tables, via intermediate Columns objects or by full
+# path
+cols['sub1/sub2/']
+cols['sub1/sub2/id']
+cols['sub1']['sub2/']['id']
+
+# Append more data to the columns. The input data is a structured
+# array or a dict of arrays.
+
+>>> cols.append(moredata1)
+>>> cols['sub1/sub2/'].append(moredata2)
+
+# add indexes for id and name
+cols['id'].create_index()
+cols['name'].create_index()
+
+# you can also create tables from a schema.  The schema itself
+# can be created from an array, individual Column schemas, or from a dict
+
+# here we use compression for id, and set the compressed chunk
+# size to 10 megabytes
+
+schema = pyc.TableSchema.from_array(array, compression=['id'], chunksize='10m')
+cols.create_table(schema=schema, name='fromschema/')
+
+# or you can build the schema from individual column schema
+cid = pyc.ColumnSchema('id', dtype='i8', compression=True)
+cname = pyc.ColumnSchema('name', dtype='i8', compression=True)
+schema = pyc.TableSchema([cid, cname])
 
-# write columns from the fields in a rec array names in the data correspond
-# to column names.  If this is the first time writing data, the columns are
-# created, and on subsequent writes, the columns must match
+# Note you specify compression info in detail
+cname = pyc.ColumnSchema(
+    'name',
+    dtype='U5',
+    compression={'cname': 'zstd', 'zlevel': 5, 'shuffle': 'bitshuffle'}
+    chunksize='10m',  # 10 megabytes
+)
 
->>> c.append(recdata)
->>> c.append(new_data)
+# You can also use a dict
+sch = {
+    'id': {
+        'dtype': 'i8',
+        'compress': True,
+    },
+    'x': {'dtype': 'f4'},
+}
 
-# append data from the fields in a FITS file
->>> c.from_fits(fitsfile_name)
+schema = pyc.TableSchema.from_schema(sch)
 
-# add a dictionary column.
->>> c.create_column('weather', 'dict')
->>> c['weather'].write({'temp': 30.1, 'humid': 0.5})
+#
+# add a new column.
+#
+
+# Default fill value is zeros, but you can set it with fill_value
+>>> cschema = pyc.ColumnSchema('newfcol', dtype='f4')
+>>> cols.create_column(cschema)
+>>> assert np.all(cols['newfcol'][:] == 0)
 
-# overwrite dict column
->>> c['weather'].write({'temp': 33.2, 'humid': 0.3, 'windspeed': 60.5})
+>>> cschema = pyc.ColumnSchema('newscol', dtype='U4', fill_value='none')
+>>> cols.create_column(cschema)
+>>> assert np.all(cols['newscol'][:] == 'none')
 
-# you should not generally create array columns, since they
-# can get out of sync with existing columns.  This will by default
-# raise an exception, but you can send verify=False if you know
-# what you are doing.  In the future special support will be added for
-# adding new columns
->>> c.create_column('test', 'array')
+# add a metadata entry
+>>> weather = {'temperature': 30, 'humidity': 50}
+>>> cols.create_meta('weather', weather)
 
-# update values for an array column
->>> c['id'][35] = 10
->>> c['id'][35:35+3] = [8, 9, 10]
->>> c['id'][rows] = idvalues
+# Update column data
+cols['x'][10:20] = 3
+cols['name'][[5, 6]] = ['alpha', 'beta']
+cols['y'][50] = 1.25
 
-# get all names, including dictionary and sub Columns
-# same as list(c.keys())
->>> c.names
-['ccd', 'dec', 'exposurename', 'id', 'imag', 'ra', 'x', 'y', 'g',
- 'meta', 'psfstars']
+# if you have indices it can be better not to rebuild them
+# until all your updates are finished. Use an updating context
+# for this: the index is not updated until exiting the context
+
+with cols.updating():
+    cols.append(newdata1)
+    cols.append(newdata2)
+
+with cols['id'].updating():
+    cols['id'][5:10] = 33
+    cols['id'][99:200] = 66
+
+# if you have updated compressed columns, you will at some point want to run
+# vaccum which defragments the files
+
+# vacuum all compressed columns
+cols.vacuum()
+
+# a specific column
+cols['id'].vacuum()
+
+# you can vacuum on exit from updating contexts
+with cols['id'].updating(vacuum=True):
+    cols['id'][5:10] = 33
+    cols['id'][99:200] = 66
+
+# You can use update to update dictionary metadata.
+# Only the fields in the input dictionary are updated or added
+>>> cols.meta['weather'].update({'temperature': 30.1, 'extra': 5})
+
+# completely overwrite
+>>> cols['weather'].write([3, 4, 5])
+
+#
+# getting names of entries
+#
+
+# get all names, including direct sub tables of the root
+>>> cols.names
+['telemetry/', 'id', 'y', 'x', 'name', 'meta']
 
 # only array column names
->>> c.column_names
-['ccd', 'dec', 'exposurename', 'id', 'imag', 'ra', 'x', 'y', 'g']
+>>> cols.column_names
+['id', 'y', 'x', 'name']
 
-# only dict columns
->>> c.dict_names
-['meta', 'weather']
+# only metadata names
+>>> cols.meta_names
+['versions', 'weather']
 
-# only sub Columns directories
->>> c.subcols_names
-['psfstars']
+# only sub table directories (only those directly above root are shown)
+>>> cols.sub_table_names
+['telemetry/']
 
 # reload all columns or specified column/column list
->>> c.reload()
+>>> cols.reload()
 
-# delete all data.  This will ask for confirmation
->>> c.delete()
+# delete a column or other entry.  This will prompt the user
+# to confirm the deletion
+>>> cols.delete_entry('id')
+# no prompt
+>>> cols.delete_entry('id', yes=True)
 
-# delete column and its data
->>> c.delete_column('ra')
+# delete all data in the columns store
+# with prompt
+>>> cols.delete()
+# no prompt
+>>> cols.delete(yes=True)
 
 # to configure the amount of memory used during index creation, specify
-# cache_mem in gigabytes.  Default 1 gig
->>> cols = pyc.Columns(fname, cache_mem=0.5)
-
-# columns can actually be another pycolumns directory
->>> psfcols = cols['psfstars']
->>> psfcols
-  dir: /some/path/mydata.cols/psfstars.cols
-  Columns:
-    name             type  dtype index  shape
-    --------------------------------------------------
-    ccd             array    <i2 True   (64348146,)
-    id              array    <i8 True   (64348146,)
-    imag            array    <f4 False  (64348146,)
-    x               array    <f4 False  (64348146,)
-    y               array    <f4 False  (64348146,)
-    ...etc
+# cache_mem is 0.5 gigabytes
+>>> cols = pyc.Columns(fname, cache_mem='0.5g')
 ```
 
-Dependencies
+Installation
 ------------
-numpy
+pip install pycolumns
 
+Dependencies
+------------
 
+pycolumns depends on numpy and [blosc](https://github.com/Blosc/python-blosc)
+for data compression. Both dependencies will be automatically installed when
+installing pycolumns with pip
diff --git a/pycolumns/__init__.py b/pycolumns/__init__.py
index 4d369db..ee9dbd9 100644
--- a/pycolumns/__init__.py
+++ b/pycolumns/__init__.py
@@ -2,14 +2,33 @@
 
 from .version import __version__
 
+from . import defaults
+from .defaults import (
+    DEFAULT_COMPRESSION,
+    DEFAULT_CNAME,
+    DEFAULT_CLEVEL,
+    DEFAULT_SHUFFLE,
+)
 from . import columns
-from .columns import Columns, where
+from .columns import Columns
+
+from . import chunks
+
+from .schema import ColumnSchema, TableSchema
 
 from . import util
+
 from . import column
-from .column import ArrayColumn, DictColumn
+from .column import Column
+
+from . import metafile
 
 from . import indices
 from .indices import Indices
 
 from . import mergesort
+
+from . import _column_pywrap
+from . import _column
+
+from . import convenience
diff --git a/pycolumns/_column.py b/pycolumns/_column.py
new file mode 100644
index 0000000..9b7be05
--- /dev/null
+++ b/pycolumns/_column.py
@@ -0,0 +1,425 @@
+import os
+import numpy as np
+from . import _column_pywrap
+from . import util
+
+
+class Column(_column_pywrap.Column):
+    """
+    A column file, with rows of fixed length records
+    """
+    def __init__(self, filename, dtype, mode='r', verbose=False):
+        self._filename = filename
+        self._dtype = np.dtype(dtype)
+        self._mode = mode
+        self._verbose = verbose
+        super().__init__(filename, mode, verbose)
+        self._set_nrows()
+
+    @property
+    def filename(self):
+        """
+        get the filename
+        """
+        return self._filename
+
+    @property
+    def mode(self):
+        """
+        get the file open mode
+        """
+        return self._mode
+
+    @property
+    def dtype(self):
+        """
+        get the numpy dtype
+        """
+        return self._dtype
+
+    @property
+    def size(self):
+        """
+        get numbe of rows
+        """
+        return self._nrows
+
+    @property
+    def nrows(self):
+        """
+        get numbe of rows
+        """
+        return self._nrows
+
+    @property
+    def verbose(self):
+        """
+        get the filename
+        """
+        return self._verbose
+
+    def resize(self, nrows):
+        """
+        Expand or truncate the file to num rows, filling with zeros if needed.
+
+        Parameters
+        ----------
+        nrows: int
+            New number of rows
+        """
+        if nrows < 0:
+            raise ValueError(f'cannot reduce rows to {nrows}')
+
+        if nrows != self.nrows:
+            nbytes = nrows * self.dtype.itemsize
+            self._resize_bytes(nbytes)
+            self._nrows = nrows
+
+    def append(self, data):
+        """
+        append data to the file
+        """
+        data = util.get_data_with_conversion(data, self.dtype)
+
+        super()._append(data)
+
+        self._nrows += data.size
+
+    def update_row(self, row, data):
+        """
+        Update the specified row
+        """
+        data = util.get_data_with_conversion(data, self.dtype)
+
+        if data.size != 1:
+            raise ValueError(
+                f'attemting to update row with data of size {data.size}'
+            )
+
+        super()._write_at(data, row)
+
+    def write_at(self, data, row):
+        """
+        write data starting at the specified row
+        """
+        data = util.get_data_with_conversion(data, self.dtype)
+
+        if row > self.nrows - 1:
+            raise IndexError(
+                f'attempt to write at row {row} > {self.nrows-1}'
+            )
+
+        super()._write_at(data, row)
+
+        if row + data.size > self.nrows:
+            self._nrows = row + data.size
+
+    def _fill_slice(self, value, s):
+        """
+        Fill the slice with the indicated value
+        """
+        data = util.get_data_with_conversion(value, self.dtype)
+        if data.size != 1:
+            raise IndexError(
+                f'cannot fill with length {data.size}'
+            )
+        super()._fill_slice(data, s.start, s.stop)
+
+    def _fill_rows(self, data, rows, sortind=None):
+        if data.size != 1:
+            raise IndexError(
+                f'cannot fill with length {data.size}'
+            )
+        if sortind is not None:
+            super()._fill_rows_sortind(data, rows, sortind)
+        else:
+            super()._fill_rows(data, rows)
+
+    def read_into(self, data):
+        """
+        read rows from the start of the file, storing in the input array
+
+        Returns
+        -------
+        data: array
+            The output data
+        """
+
+        self._check_dtype(data)
+
+        nrows = self.nrows
+        if data.size > nrows:
+            raise ValueError(
+                f'input data rows {data.size} > file nrows {nrows}'
+            )
+        super()._read_slice(data, 0)
+
+    def read_slice_into(self, data, s):
+        """
+        read rows into the input data
+
+        Parameters
+        ----------
+        data: array
+            The slice.  Must have a start
+        s: slice
+            The slice.  Must have a start and no step
+
+        Returns
+        -------
+        None
+        """
+
+        self._check_dtype(data)
+
+        s = util.extract_slice(s, self.nrows)
+        nrows = s.stop - s.start
+        if data.size != nrows:
+            raise ValueError(f'data size {data.size} != slice nrows {nrows}')
+        super()._read_slice(data, s.start)
+
+    def read_rows_into(self, data, rows):
+        """
+        read rows into the input data
+
+        Parameters
+        ----------
+        data: array
+            Array in which to store the values
+        rows: array
+            The rows array
+
+        Returns
+        -------
+        None
+        """
+        from .indices import Indices
+        self._check_dtype(data)
+
+        rows_use = util.extract_rows(rows, self.nrows)
+        if not isinstance(rows_use, Indices):
+            raise ValueError(f'git unexpected rows type {type(rows_use)}')
+
+        super()._read_rows(data, rows_use)
+        # super()._read_rows_pages(data, rows)
+
+    def read_row_into(self, data, row):
+        """
+        read rows into the input data
+
+        Parameters
+        ----------
+        data: array
+            Array in which to store the values. Even though reading
+            a single row, still must be length one array not scalar
+        rows: array
+            The rows array
+
+        Returns
+        -------
+        None
+        """
+        self._check_dtype(data)
+        if data.ndim == 0:
+            raise ValueError('data must have ndim > 0')
+
+        super()._read_row(data, row)
+        # super()._read_row_pages(data, rows)
+
+    def __getitem__(self, arg):
+
+        # returns either Indices or slice
+        # converts slice with step to indices
+        rows = util.extract_rows(arg, self.nrows)
+
+        if isinstance(rows, slice):
+            nrows = rows.stop - rows.start
+            data = np.empty(nrows, dtype=self.dtype)
+            super()._read_slice(data, rows.start)
+        else:
+            data = np.empty(rows.size, dtype=self.dtype)
+
+            if rows.ndim == 0:
+                self._read_row(data, rows)
+                data = data[0]
+            else:
+                sortind = rows.sort_index
+                if sortind is None:
+                    self._read_rows(data, rows)
+                else:
+                    self._read_rows_sortind(data, rows, sortind)
+
+        return data
+
+    def __setitem__(self, arg, data):
+        # returns either Indices or slice
+        # converts slice with step to indices
+        rows = util.extract_rows(arg, self.nrows, check_slice_stop=True)
+
+        if isinstance(rows, slice):
+            nrows = rows.stop - rows.start
+            if np.isscalar(data) or (len(data) == 1 and nrows > 1):
+                self._fill_slice(data, rows)
+            else:
+                if nrows != len(data):
+                    raise IndexError(
+                        f'mismatch slice size {nrows} and data '
+                        f'size {len(data)} when writing'
+                    )
+
+                self.write_at(data, rows.start)
+        else:
+            if rows.ndim == 0:
+                self.write_at(data, rows)
+            else:
+                data = util.get_data_with_conversion(data, self.dtype)
+
+                sortind = rows.sort_index
+
+                # rows are sorted, can check first and last
+                if sortind is None:
+                    first = rows[0]
+                    last = rows[-1]
+                else:
+                    first = rows[sortind[0]]
+                    last = rows[sortind[-1]]
+
+                self._check_row(first)
+                self._check_row(last)
+
+                if data.size == 1 and rows.size > 1:
+                    self._fill_rows(data, rows, sortind=sortind)
+                else:
+                    if sortind is None:
+                        self._write_rows(data, rows)
+                    else:
+                        self._write_rows_sortind(data, rows, sortind)
+
+    def _check_row(self, row):
+        if row < 0 or row > self.nrows - 1:
+            raise IndexError(
+                f'row {row} out of bounds [0, {self.nrows-1}]'
+            )
+
+    def _check_dtype(self, data):
+        dtype = data.dtype
+        if dtype != self.dtype:
+            raise ValueError(
+                f'input dtype {dtype} != file dtype {self.dtype}'
+            )
+
+    def _set_nrows(self):
+        fsize = os.path.getsize(self.filename)
+        elsize = self.dtype.itemsize
+
+        if fsize % elsize != 0:
+            raise ValueError(
+                f'file size {fsize} not a multiple of element size {elsize}'
+            )
+        self._nrows = fsize // elsize
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, exception_type, exception_value, traceback):
+        self.close()
+
+    def __repr__(self):
+        indent = '    '
+
+        if self.dtype.names is not None:
+            dt = self.dtype.descr
+        else:
+            dt = self.dtype.str
+
+        rep = [f'filename: {self.filename}']
+        rep += [f'mode: {self.mode}']
+        rep += [f'dtype: {dt}']
+        rep += [f'nrows: {self.nrows}']
+
+        rep = [indent + r for r in rep]
+
+        rep = ['Column:'] + rep
+        return '\n'.join(rep)
+
+
+def read(fname, dtype, rows=None, verbose=False):
+    with Column(fname, dtype=dtype, mode='r', verbose=verbose) as col:
+        if rows is None:
+            data = col[:]
+        else:
+            data = col[rows]
+    return data
+
+
+def write(fname, data, append=True, verbose=False):
+    if os.path.exists(fname) and append:
+        mode = 'r+'
+    else:
+        mode = 'w+'
+
+    with Column(fname, dtype=data.dtype, mode=mode, verbose=verbose) as col:
+        if not col.has_header():
+            col.init(data.dtype.str)
+        col.append(data)
+
+
+def append(fname, data, verbose=False):
+    if not os.path.exists(fname):
+        mode = 'w+'
+    else:
+        mode = 'r+'
+
+    with Column(fname, dtype=data.dtype, mode=mode, verbose=verbose) as col:
+        if not col.has_header():
+            col.init(data.dtype.str)
+        col.append(data)
+
+
+def test():
+    import tempfile
+    with tempfile.TemporaryDirectory() as tmpdir:
+        data = np.arange(20)
+        fname = os.path.join(tmpdir, 'test.col')
+        with Column(fname, dtype=data.dtype, mode='w+', verbose=True) as col:
+            print('-' * 70)
+            print('before append')
+            print(col)
+
+            col.append(data)
+
+            print('-' * 70)
+            print('before append')
+            print(col)
+
+            indata = col[:]
+            assert np.all(indata == data)
+
+            indata = col[2:8]
+            assert np.all(indata == data[2:8])
+
+            indata = col[2:18:2]
+            assert np.all(indata == data[2:18:2])
+
+            ind = [3, 5, 7]
+            indata = col[ind]
+            assert np.all(indata == data[ind])
+
+            ind = 5
+            indata = col[ind]
+            assert np.all(indata == data[ind])
+
+            s = slice(2, 8)
+            indata = np.zeros(s.stop - s.start, dtype=data.dtype)
+            col.read_slice_into(indata, s)
+            assert np.all(indata == data[s])
+
+            ind = [3, 5, 7]
+            indata = np.zeros(len(ind), dtype=data.dtype)
+            col.read_rows_into(indata, ind)
+            assert np.all(indata == data[ind])
+
+            # ind = [3, 5, 7]
+            # indata = np.zeros(5, dtype=data.dtype)
+            # col.read_rows_into(indata, ind)
+            # assert np.all(indata == data[ind])
diff --git a/pycolumns/_column_pywrap.c b/pycolumns/_column_pywrap.c
new file mode 100644
index 0000000..2aa5b0e
--- /dev/null
+++ b/pycolumns/_column_pywrap.c
@@ -0,0 +1,948 @@
+/*
+   TODO
+
+   - python wrapper class or inherited
+   - read one without checking erros for speed?  Or is read slice OK?
+   - maybe more error checking internally
+   - printouts when verbose
+*/
+#define NPY_NO_DEPRECATED_API NPY_1_7_API_VERSION
+
+// #include <string.h>
+// #include <math.h>
+#include <Python.h>
+#include "numpy/arrayobject.h" 
+
+struct PyColumn {
+    PyObject_HEAD
+
+    const char* fname;
+    const char* mode;
+    int verbose;
+    FILE* fptr;
+
+};
+
+static int
+ensure_arrays_same_size(
+  PyArrayObject* arr1, const char* name1,
+  PyArrayObject* arr2, const char* name2
+  ) {
+    npy_intp num1 = 0, num2 = 0;
+    num1 = PyArray_SIZE(arr1);
+    num2 = PyArray_SIZE(arr2);
+
+    if (num1 != num2) {
+        PyErr_Format(
+            PyExc_ValueError,
+            "%s has size %" NPY_INTP_FMT
+            " but %s has size %" NPY_INTP_FMT, name1, num1, name2, num2
+        );
+        return 0;
+    } else {
+        return 1;
+    }
+}
+
+static int
+check_array_size(PyArrayObject* array, npy_intp expected) {
+    npy_intp num = 0;
+    num = PyArray_SIZE(array);
+
+    if (num != expected) {
+        PyErr_Format(
+            PyExc_ValueError,
+            "array has size %" NPY_INTP_FMT
+            " but expected %" NPY_INTP_FMT, num, expected
+        );
+        return 0;
+    } else {
+        return 1;
+    }
+}
+
+static npy_intp
+pyc_get_row_offset(npy_intp row, npy_intp elsize) {
+    return row * elsize;
+}
+
+static void
+pyc_seek_row(struct PyColumn* self, npy_intp row, npy_intp elsize) {
+    npy_intp offset = 0;
+    offset = pyc_get_row_offset(row, elsize);
+    fseek(self->fptr, offset, SEEK_SET);
+}
+
+// Resize the file to nbytes, expanding with zeros if necessary
+static PyObject*
+PyColumn_resize_bytes(
+    struct PyColumn* self,
+    PyObject *args,
+    PyObject *kwds
+)
+{
+    long long nbytes = 0;
+
+    if (!PyArg_ParseTuple(args, (char*)"L", &nbytes)) {
+        return NULL;
+    }
+
+    rewind(self->fptr);
+    if (0 != ftruncate(fileno(self->fptr), nbytes)) {
+        PyErr_Format(PyExc_IOError,
+                     "Could not truncate file to %lld bytes: %s",
+                     nbytes, strerror(errno));
+        return NULL;
+    }
+    fflush(self->fptr);
+
+    Py_RETURN_NONE;
+}
+
+// Fill a slice with a value
+static PyObject*
+PyColumn_fill_slice(
+    struct PyColumn* self,
+    PyObject *args,
+    PyObject *kwds
+)
+{
+    long long start = 0, stop = 0;
+    npy_intp elsize = 0, row = 0, nwrote = 0;
+    PyArrayObject* array = NULL;
+    void *ptr = NULL;
+
+    if (!PyArg_ParseTuple(args, (char*)"OLL", &array, &start, &stop)) {
+        return NULL;
+    }
+
+    if (!check_array_size(array, 1)) {
+        return NULL;
+    }
+
+    elsize = PyArray_ITEMSIZE(array);
+    ptr = PyArray_GETPTR1(array, 0);
+
+    for (row=start; row < stop; row++) {
+        pyc_seek_row(self, row, elsize);
+
+        nwrote = fwrite(
+            ptr,
+            elsize,
+            1,
+            self->fptr
+        );
+        if (nwrote < 1) {
+            // NPY_END_THREADS;
+            PyErr_Format(PyExc_IOError,
+                         "Error writing row %" NPY_INTP_FMT
+                         " to file", row);
+            return NULL;
+        }
+
+        /* if (PyArray_ToFile(array, self->fptr, "", "") != 0) { */
+        /*     PyErr_Format(PyExc_IOError, */
+        /*                  "Error writing data to %s with mode %s", */
+        /*                  self->fname, self->mode); */
+        /*     return NULL; */
+        /* } */
+    }
+
+    fflush(self->fptr);
+
+    Py_RETURN_NONE;
+}
+
+
+// Append data to file
+static PyObject*
+PyColumn_append(
+    struct PyColumn* self,
+    PyObject *args,
+    PyObject *kwds
+)
+{
+    PyArrayObject* array = NULL;
+
+    if (!PyArg_ParseTuple(args, (char*)"O", &array)) {
+        return NULL;
+    }
+
+    fseek(self->fptr, 0, SEEK_END);
+    if (PyArray_ToFile(array, self->fptr, "", "") != 0) {
+        PyErr_Format(PyExc_IOError,
+                     "Error writing data to %s with mode %s",
+                     self->fname, self->mode);
+        return NULL;
+    }
+    fflush(self->fptr);
+
+    Py_RETURN_NONE;
+}
+
+// write data from the specified row location
+static PyObject*
+PyColumn_write_at(
+    struct PyColumn* self,
+    PyObject *args,
+    PyObject *kwds
+)
+{
+    PyArrayObject* array = NULL;
+    long long start = 0;
+    npy_intp elsize = 0;
+
+    if (!PyArg_ParseTuple(args, (char*)"OL", &array, &start)) {
+        return NULL;
+    }
+
+    elsize = PyArray_ITEMSIZE(array);
+    pyc_seek_row(self, start, elsize);
+
+    if (PyArray_ToFile(array, self->fptr, "", "") != 0) {
+        PyErr_Format(PyExc_IOError,
+                     "Error writing data to %s with mode %s",
+                     self->fname, self->mode);
+        return NULL;
+    }
+    fflush(self->fptr);
+
+    Py_RETURN_NONE;
+}
+
+/*
+   Write rows into file
+   No error checking is done on the data type of the input array
+   The rows should be sorted for efficiency, but this is not checked
+   rows should be native npy_int64 but this is not checked
+*/
+
+static PyObject*
+PyColumn_write_rows(
+    struct PyColumn* self,
+    PyObject *args,
+    PyObject *kwds
+)
+{
+    PyArrayObject* array = NULL, *rows = NULL;
+    npy_int64 row = 0, elsize = 0;
+    npy_intp nwrote = 0;
+    PyArrayIterObject *it = NULL;
+    // NPY_BEGIN_THREADS_DEF;
+
+    if (!PyArg_ParseTuple(args, (char*)"OO", &array, &rows)) {
+        return NULL;
+    }
+
+    if (!ensure_arrays_same_size(rows, "rows", array, "array")) {
+        return NULL;
+    }
+
+    elsize = PyArray_ITEMSIZE(array);
+    it = (PyArrayIterObject *) PyArray_IterNew((PyObject *)array);
+
+    // NPY_BEGIN_THREADS;
+
+    while (it->index < it->size) {
+        row = *(npy_int64 *) PyArray_GETPTR1(rows, it->index);
+
+        pyc_seek_row(self, row, elsize);
+
+        nwrote = fwrite(
+            (void *) it->dataptr,
+            elsize,
+            1,
+            self->fptr
+        );
+        if (nwrote < 1) {
+            // NPY_END_THREADS;
+            PyErr_Format(PyExc_IOError,
+                         "Error writing row %" NPY_INTP_FMT
+                         " to file", row);
+            Py_DECREF(it);
+            return NULL;
+        }
+        PyArray_ITER_NEXT(it);
+    }
+    // NPY_END_THREADS;
+    Py_DECREF(it);
+
+    Py_RETURN_NONE;
+}
+
+/*
+   Write rows into file with a sort index
+   No error checking is done on the data type of the input array
+   The rows should be sorted for efficiency, but this is not checked
+   rows should be native npy_int64 but this is not checked
+*/
+
+static PyObject*
+PyColumn_write_rows_sortind(
+    struct PyColumn* self,
+    PyObject *args,
+    PyObject *kwds
+)
+{
+    PyArrayObject* array = NULL, *rows = NULL, *sortind = NULL;
+    npy_int64 row = 0, s = 0, elsize = 0, index = 0, size = 0;
+    npy_intp nwrote = 0;
+    void *ptr = NULL;
+    // NPY_BEGIN_THREADS_DEF;
+
+    if (!PyArg_ParseTuple(args, (char*)"OOO", &array, &rows, &sortind)) {
+        return NULL;
+    }
+
+    if (!ensure_arrays_same_size(rows, "rows", array, "array")) {
+        return NULL;
+    }
+    if (!ensure_arrays_same_size(rows, "rows", sortind, "sortind")) {
+        return NULL;
+    }
+
+    size = PyArray_SIZE(array);
+    elsize = PyArray_ITEMSIZE(array);
+
+    // NPY_BEGIN_THREADS;
+
+    for (index = 0; index < size; index ++) {
+        s = *(npy_int64 *) PyArray_GETPTR1(sortind, index);
+        ptr = PyArray_GETPTR1(array, s);
+        row = *(npy_int64 *) PyArray_GETPTR1(rows, s);
+
+        pyc_seek_row(self, row, elsize);
+
+        nwrote = fwrite(
+            ptr,
+            elsize,
+            1,
+            self->fptr
+        );
+        if (nwrote < 1) {
+            // NPY_END_THREADS;
+            PyErr_Format(PyExc_IOError,
+                         "Error writing row %" NPY_INTP_FMT
+                         " to file", row);
+            return NULL;
+        }
+    }
+    // NPY_END_THREADS;
+
+    Py_RETURN_NONE;
+}
+
+/*
+   Fill rows with the input value
+*/
+
+static PyObject*
+PyColumn_fill_rows(
+    struct PyColumn* self,
+    PyObject *args,
+    PyObject *kwds
+)
+{
+    PyArrayObject* array = NULL, *rows = NULL;
+    npy_int64 row = 0, elsize = 0;
+    npy_intp nwrote = 0;
+    PyArrayIterObject *it = NULL;
+    void *ptr = NULL;
+
+    // NPY_BEGIN_THREADS_DEF;
+
+    if (!PyArg_ParseTuple(args, (char*)"OO", &array, &rows)) {
+        return NULL;
+    }
+
+    if (!check_array_size(array, 1)) {
+        return NULL;
+    }
+
+    elsize = PyArray_ITEMSIZE(array);
+    ptr = PyArray_GETPTR1(array, 0);
+
+    it = (PyArrayIterObject *) PyArray_IterNew((PyObject *)rows);
+
+    // NPY_BEGIN_THREADS;
+
+    while (it->index < it->size) {
+        row = *(npy_int64 *) it->dataptr;
+
+        pyc_seek_row(self, row, elsize);
+
+        nwrote = fwrite(
+            ptr,
+            elsize,
+            1,
+            self->fptr
+        );
+        if (nwrote < 1) {
+            // NPY_END_THREADS;
+            PyErr_Format(PyExc_IOError,
+                         "Error writing row %" NPY_INTP_FMT
+                         " to file", row);
+            Py_DECREF(it);
+            return NULL;
+        }
+        PyArray_ITER_NEXT(it);
+    }
+    // NPY_END_THREADS;
+    Py_DECREF(it);
+
+    Py_RETURN_NONE;
+}
+
+/*
+   Fill rows with the input value, using a sort index
+*/
+
+static PyObject*
+PyColumn_fill_rows_sortind(
+    struct PyColumn* self,
+    PyObject *args,
+    PyObject *kwds
+)
+{
+    PyArrayObject* array = NULL, *rows = NULL, *sortind = NULL;
+    npy_int64 row = 0, s = 0, elsize = 0, index = 0, size = 0;
+    npy_intp nwrote = 0;
+    void *ptr = NULL;
+    // NPY_BEGIN_THREADS_DEF;
+
+    if (!PyArg_ParseTuple(args, (char*)"OOO", &array, &rows, &sortind)) {
+        return NULL;
+    }
+
+    if (!check_array_size(array, 1)) {
+        return NULL;
+    }
+    if (!ensure_arrays_same_size(rows, "rows", sortind, "sortind")) {
+        return NULL;
+    }
+
+    size = PyArray_SIZE(rows);
+    elsize = PyArray_ITEMSIZE(array);
+    ptr = PyArray_GETPTR1(array, 0);
+
+    // NPY_BEGIN_THREADS;
+
+    for (index = 0; index < size; index ++) {
+        s = *(npy_int64 *) PyArray_GETPTR1(sortind, index);
+        row = *(npy_int64 *) PyArray_GETPTR1(rows, s);
+
+        pyc_seek_row(self, row, elsize);
+
+        nwrote = fwrite(
+            ptr,
+            elsize,
+            1,
+            self->fptr
+        );
+        if (nwrote < 1) {
+            // NPY_END_THREADS;
+            PyErr_Format(PyExc_IOError,
+                         "Error writing row %" NPY_INTP_FMT
+                         " to file", row);
+            return NULL;
+        }
+    }
+    // NPY_END_THREADS;
+
+    Py_RETURN_NONE;
+}
+
+
+/*
+   Read slice into array
+   No error checking is done on the data type of the input array
+*/
+static PyObject*
+PyColumn_read_slice(
+    struct PyColumn* self,
+    PyObject *args,
+    PyObject *kwds
+)
+{
+    PyArrayObject* array = NULL;
+    long long start = 0;
+    npy_intp num = 0, row = 0, nread = 0, elsize = 0;
+    PyArrayIterObject *it = NULL;
+
+    if (!PyArg_ParseTuple(args, (char*)"OL", &array, &start)) {
+        return NULL;
+    }
+
+    elsize = PyArray_ITEMSIZE(array);
+    num = PyArray_SIZE(array);
+
+    pyc_seek_row(self, start, elsize);
+
+    if (PyArray_ISCONTIGUOUS(array)) {
+        if (self->verbose) {
+             fprintf(stderr, "reading as contiguous\n");
+        }
+        // NPY_BEGIN_ALLOW_THREADS;
+        nread = fread(PyArray_DATA(array),
+                      elsize,
+                      num,
+                      self->fptr);
+        // NPY_END_ALLOW_THREADS;
+        if (nread != num) {
+            PyErr_Format(PyExc_IOError,
+                "Error reading slice [%" NPY_INTP_FMT ", %" NPY_INTP_FMT ") "
+                "from %s", start, start+num, self->fname);
+            return NULL;
+        }
+    } else {
+        if (self->verbose) {
+            fprintf(stderr, "reading as non contiguous\n");
+        }
+        // NPY_BEGIN_THREADS_DEF;
+
+        it = (PyArrayIterObject *) PyArray_IterNew((PyObject *)array);
+
+        // NPY_BEGIN_THREADS;
+
+        row = start;
+        while (it->index < it->size) {
+            nread = fread(
+                (void *) it->dataptr,
+                elsize,
+                1,
+                self->fptr 
+            );
+            if (nread < 1) {
+                // NPY_END_THREADS;
+                PyErr_Format(PyExc_IOError,
+                             "Error reading row %" NPY_INTP_FMT
+                             " from file", row);
+                Py_DECREF(it);
+                return NULL;
+            }
+            row += 1;
+            PyArray_ITER_NEXT(it);
+        }
+        // NPY_END_THREADS;
+        Py_DECREF(it);
+
+    }
+
+    Py_RETURN_NONE;
+}
+
+/*
+   Read rows into array
+   No error checking is done on the data type of the input array
+   The rows should be sorted for efficiency, but this is not checked
+   rows should be native npy_int64 but this is not checked
+*/
+
+static PyObject*
+PyColumn_read_rows(
+    struct PyColumn* self,
+    PyObject *args,
+    PyObject *kwds
+)
+{
+    PyArrayObject* array = NULL, *rows = NULL;
+    npy_int64 row = 0, elsize = 0;
+    npy_intp nread = 0;
+    PyArrayIterObject *it = NULL;
+    // NPY_BEGIN_THREADS_DEF;
+
+    if (!PyArg_ParseTuple(args, (char*)"OO", &array, &rows)) {
+        return NULL;
+    }
+
+    if (!ensure_arrays_same_size(rows, "rows", array, "array")) {
+        return NULL;
+    }
+
+    elsize = PyArray_ITEMSIZE(array);
+    it = (PyArrayIterObject *) PyArray_IterNew((PyObject *)array);
+
+    // NPY_BEGIN_THREADS;
+
+    while (it->index < it->size) {
+        row = *(npy_int64 *) PyArray_GETPTR1(rows, it->index);
+
+        pyc_seek_row(self, row, elsize);
+
+        nread = fread(
+            (void *) it->dataptr,
+            elsize,
+            1,
+            self->fptr 
+        );
+        if (nread < 1) {
+            // NPY_END_THREADS;
+            PyErr_Format(PyExc_IOError,
+                         "Error reading row %" NPY_INTP_FMT
+                         " from file", row);
+            Py_DECREF(it);
+            return NULL;
+        }
+        PyArray_ITER_NEXT(it);
+    }
+    // NPY_END_THREADS;
+    Py_DECREF(it);
+
+    Py_RETURN_NONE;
+}
+
+/*
+   Read rows into array with sort indices
+   No error checking is done on the data type of the input array
+   The rows should be sorted for efficiency, but this is not checked
+   rows should be native npy_int64 but this is not checked
+*/
+
+static PyObject*
+PyColumn_read_rows_sortind(
+    struct PyColumn* self,
+    PyObject *args,
+    PyObject *kwds
+)
+{
+    PyArrayObject* array = NULL, *rows = NULL, *sortind = NULL;
+    npy_int64 row = 0, s = 0, elsize = 0, index = 0, size = 0;
+    npy_intp nread = 0;
+    // NPY_BEGIN_THREADS_DEF;
+    void *ptr = NULL;
+
+    if (!PyArg_ParseTuple(args, (char*)"OOO", &array, &rows, &sortind)) {
+        return NULL;
+    }
+
+    if (!ensure_arrays_same_size(rows, "rows", array, "array")) {
+        return NULL;
+    }
+    if (!ensure_arrays_same_size(rows, "rows", sortind, "sortind")) {
+        return NULL;
+    }
+
+    size = PyArray_SIZE(array);
+    elsize = PyArray_ITEMSIZE(array);
+
+    // NPY_BEGIN_THREADS;
+
+    for (index = 0; index < size; index ++) {
+        s = *(npy_int64 *) PyArray_GETPTR1(sortind, index);
+        ptr = PyArray_GETPTR1(array, s);
+        row = *(npy_int64 *) PyArray_GETPTR1(rows, s);
+
+        pyc_seek_row(self, row, elsize);
+
+        nread = fread(
+            ptr,
+            elsize,
+            1,
+            self->fptr 
+        );
+        if (nread < 1) {
+            // NPY_END_THREADS;
+            PyErr_Format(PyExc_IOError,
+                         "Error reading row %" NPY_INTP_FMT
+                         " from file", row);
+            return NULL;
+        }
+    }
+    // NPY_END_THREADS;
+
+    Py_RETURN_NONE;
+}
+
+
+static PyObject*
+PyColumn_read_row(
+    struct PyColumn* self,
+    PyObject *args,
+    PyObject *kwds
+)
+{
+    PyArrayObject* array = NULL;
+    long long row = 0;
+    npy_intp nread = 0, elsize = 0;
+
+    if (!PyArg_ParseTuple(args, (char*)"OL", &array, &row)) {
+        return NULL;
+    }
+
+    if (!check_array_size(array, 1)) {
+        return NULL;
+    }
+
+    elsize = PyArray_ITEMSIZE(array);
+
+    pyc_seek_row(self, row, elsize);
+
+    nread = fread(
+        (void *) PyArray_GETPTR1(array, 0),
+        elsize,
+        1,
+        self->fptr 
+    );
+
+    if (nread < 1) {
+        PyErr_Format(PyExc_IOError,
+                     "Error reading row %" NPY_INTP_FMT
+                     " from file", row);
+        return NULL;
+    }
+
+    Py_RETURN_NONE;
+}
+
+static int
+PyColumn_init(struct PyColumn* self, PyObject *args, PyObject *kwds)
+{
+    char* fname = NULL;
+    char* mode = NULL;
+    int verbose = 0;
+
+    if (!PyArg_ParseTuple(args, (char*)"ssi", &fname, &mode, &verbose)) {
+        return -1;
+    }
+
+    self->fname = (char *) fname;
+    self->mode = (char *) mode;
+    self->verbose = verbose;
+
+    self->fptr = fopen(self->fname, self->mode);
+    if (!self->fptr) {
+        PyErr_Format(PyExc_IOError, "Error opening %s with mode %s",
+                     self->fname, self->mode);
+        return -1;
+    }
+
+    return 0;
+}
+
+static PyObject *
+PyColumn_close(struct PyColumn* self)
+{
+    if (self->fptr != NULL) {
+        fclose(self->fptr);
+        self->fptr = NULL;
+    }
+    Py_RETURN_NONE;
+}
+
+
+
+static void
+PyColumn_dealloc(struct PyColumn* self)
+{
+
+    if (self->fptr != NULL) {
+        fclose(self->fptr);
+        self->fptr = NULL;
+    }
+    Py_TYPE(self)->tp_free((PyObject*)self);
+}
+
+static PyObject *
+PyColumn_repr(struct PyColumn* self) {
+    char buff[4096];
+  
+    snprintf(buff, 4096,
+             "column\n"
+             "    file: %s\n"
+             "    mode: %s\n"
+             "    verbose: %d\n",
+             self->fname, self->mode, self->verbose);
+
+    return PyUnicode_FromString((const char*)buff);
+}
+
+static PyMethodDef PyColumn_methods[] = {
+    {"_resize_bytes",
+     (PyCFunction)PyColumn_resize_bytes,
+     METH_VARARGS, 
+     "_resize_bytes()\n"
+     "\n"
+     "Resize the file to nbytes, expanding and filling with zeros if needed"},
+
+    {"_fill_slice",
+     (PyCFunction)PyColumn_fill_slice,
+     METH_VARARGS, 
+     "_fill_slice()\n"
+     "\n"
+     "Fill the slice with the give value."},
+
+    {"_append",
+     (PyCFunction)PyColumn_append,
+     METH_VARARGS, 
+     "_append()\n"
+     "\n"
+     "Append data.\n"},
+
+    {"_write_at",
+     (PyCFunction)PyColumn_write_at,
+     METH_VARARGS, 
+     "_write_at()\n"
+     "\n"
+     "Write data from the specified row.\n"},
+
+    {"_write_rows",
+     (PyCFunction)PyColumn_write_rows,
+     METH_VARARGS, 
+     "_write_rows()\n"
+     "\n"
+     "Write data at the specified row.\n"},
+
+    {"_write_rows_sortind",
+     (PyCFunction)PyColumn_write_rows_sortind,
+     METH_VARARGS, 
+     "_write_rows_sortind()\n"
+     "\n"
+     "Write data at the specified row.\n"},
+
+    {"_fill_rows",
+     (PyCFunction)PyColumn_fill_rows,
+     METH_VARARGS, 
+     "_fill_rows()\n"
+     "\n"
+     "Fill rows with the input value.\n"},
+
+    {"_fill_rows_sortind",
+     (PyCFunction)PyColumn_fill_rows_sortind,
+     METH_VARARGS, 
+     "_fill_rows_sortind()\n"
+     "\n"
+     "Fill rows with the input value using a sort index.\n"},
+
+    {"_read_slice",
+     (PyCFunction)PyColumn_read_slice,
+     METH_VARARGS, 
+     "_read_slice()\n"
+     "\n"
+     "Read slice of data into input array.\n"},
+
+    {"_read_rows",
+     (PyCFunction)PyColumn_read_rows,
+     METH_VARARGS, 
+     "_read_rows()\n"
+     "\n"
+     "Read rows into input array.\n"},
+
+    {"_read_rows_sortind",
+     (PyCFunction)PyColumn_read_rows_sortind,
+     METH_VARARGS, 
+     "_read_rows_sortind()\n"
+     "\n"
+     "Read rows into input array using sort indices.\n"},
+
+    /*
+    {"_read_rows_pages",
+     (PyCFunction)PyColumn_read_rows_pages,
+     METH_VARARGS, 
+     "_read_rows_pages()\n"
+     "\n"
+     "Read rows into input array.\n"},
+     */
+
+    {"_read_row",
+     (PyCFunction)PyColumn_read_row,
+     METH_VARARGS, 
+     "_read_row()\n"
+     "\n"
+     "Read single row into input array.\n"},
+
+    {"close",
+     (PyCFunction)PyColumn_close,
+     METH_VARARGS, 
+     "close()\n"
+     "\n"
+     "close the file stream.\n"},
+
+    {NULL}  /* Sentinel */
+};
+
+static PyTypeObject PyColumnType = {
+    PyVarObject_HEAD_INIT(NULL, 0)
+    "_column.Column",             /*tp_name*/
+    sizeof(struct PyColumn), /*tp_basicsize*/
+    0,                         /*tp_itemsize*/
+    (destructor)PyColumn_dealloc, /*tp_dealloc*/
+    0,                         /*tp_print*/
+    0,                         /*tp_getattr*/
+    0,                         /*tp_setattr*/
+    0,                         /*tp_compare*/
+    //0,                         /*tp_repr*/
+    (reprfunc)PyColumn_repr,                         /*tp_repr*/
+    0,                         /*tp_as_number*/
+    0,                         /*tp_as_sequence*/
+    0,                         /*tp_as_mapping*/
+    0,                         /*tp_hash */
+    0,                         /*tp_call*/
+    0,                         /*tp_str*/
+    0,                         /*tp_getattro*/
+    0,                         /*tp_setattro*/
+    0,                         /*tp_as_buffer*/
+    Py_TPFLAGS_DEFAULT | Py_TPFLAGS_BASETYPE, /*tp_flags*/
+    "A class to work with rows of data.\n",
+    0,                     /* tp_traverse */
+    0,                     /* tp_clear */
+    0,                     /* tp_richcompare */
+    0,                     /* tp_weaklistoffset */
+    0,                     /* tp_iter */
+    0,                     /* tp_iternext */
+    PyColumn_methods,             /* tp_methods */
+    0,             /* tp_members */
+    0,                         /* tp_getset */
+    0,                         /* tp_base */
+    0,                         /* tp_dict */
+    0,                         /* tp_descr_get */
+    0,                         /* tp_descr_set */
+    0,                         /* tp_dictoffset */
+    //0,     /* tp_init */
+    (initproc)PyColumn_init,      /* tp_init */
+    0,                         /* tp_alloc */
+    PyType_GenericNew,                 /* tp_new */
+};
+
+static PyMethodDef column_methods[] = {
+    {NULL}  /* Sentinel */
+};
+
+
+static struct PyModuleDef moduledef = {
+    PyModuleDef_HEAD_INIT,
+    "_column_pywrap",      /* m_name */
+    "Defines the Column class",  /* m_doc */
+    -1,                  /* m_size */
+    column_methods,    /* m_methods */
+    NULL,                /* m_reload */
+    NULL,                /* m_traverse */
+    NULL,                /* m_clear */
+    NULL,                /* m_free */
+};
+
+#ifndef PyMODINIT_FUNC  /* declarations for DLL import/export */
+#define PyMODINIT_FUNC void
+#endif
+PyMODINIT_FUNC
+PyInit__column_pywrap(void) 
+{
+    PyObject* m;
+
+
+    PyColumnType.tp_new = PyType_GenericNew;
+
+    if (PyType_Ready(&PyColumnType) < 0) {
+        return NULL;
+    }
+    m = PyModule_Create(&moduledef);
+    if (m==NULL) {
+        return NULL;
+    }
+
+    Py_INCREF(&PyColumnType);
+    PyModule_AddObject(m, "Column", (PyObject *)&PyColumnType);
+
+    import_array();
+
+    return m;
+}
diff --git a/pycolumns/chunks.py b/pycolumns/chunks.py
new file mode 100644
index 0000000..1ef3d07
--- /dev/null
+++ b/pycolumns/chunks.py
@@ -0,0 +1,1054 @@
+import os
+import numpy as np
+from ._column import Column as CColumn
+from . import util
+from .defaults import CHUNKS_DTYPE
+
+
+class Chunks(object):
+    """
+    A chunked file, with possible compression
+    """
+    def __init__(
+        self,
+        filename,
+        chunks_filename,
+        dtype,
+        mode='r',
+        compression=None,
+        chunksize='1m',
+        verbose=False,
+    ):
+        self._filename = filename
+        self._chunks_filename = chunks_filename
+        self._dir = os.path.dirname(filename)
+
+        self._dtype = np.dtype(dtype)
+        self._chunks_dtype = np.dtype(CHUNKS_DTYPE)
+
+        self._set_compression(compression)
+        self._mode = mode
+        self._verbose = verbose
+        self._open_files()
+        self._set_nrows()
+        self._set_row_chunksize(chunksize)
+
+        self._clear_chunk_cache()
+
+    @property
+    def dir(self):
+        """
+        get the directory
+        """
+        return self._dir
+
+    @property
+    def filename(self):
+        """
+        get the filename
+        """
+        return self._filename
+
+    @property
+    def chunks_filename(self):
+        """
+        get the filename
+        """
+        return self._chunks_filename
+
+    @property
+    def mode(self):
+        """
+        get the file open mode
+        """
+        return self._mode
+
+    @property
+    def compression(self):
+        """
+        get the file open mode
+        """
+        return self._compression
+
+    @property
+    def dtype(self):
+        """
+        get the numpy dtype
+        """
+        return self._dtype
+
+    @property
+    def chunks_dtype(self):
+        """
+        get the numpy dtype of the chunks file data
+        """
+        return self._chunks_dtype
+
+    @property
+    def size(self):
+        """
+        get numbe of rows
+        """
+        return self._nrows
+
+    @property
+    def nrows(self):
+        """
+        get number of rows
+        """
+        return self._nrows
+
+    @property
+    def chunk_data(self):
+        """
+        get number of chunks
+        """
+        return self._chunk_data
+
+    @property
+    def nchunks(self):
+        """
+        get number of chunks
+        """
+        cd = self.chunk_data
+
+        nchunks = 0
+        if cd.has_data():
+            nchunks = cd.size
+
+        return nchunks
+
+    @property
+    def chunksize(self):
+        """
+        get string chunk size
+        """
+        return self._chunksize
+
+    @property
+    def row_chunksize(self):
+        """
+        get chunk size in rows
+        """
+        return self._row_chunksize
+
+    @property
+    def verbose(self):
+        """
+        get the filename
+        """
+        return self._verbose
+
+    def close(self):
+        """
+        close the data and chunks files
+        """
+        self._fobj.close()
+
+    def append(self, data):
+        """
+        Append data to the file. If the last chunk is not filled it
+        will be filled before appending in a new chunk
+        """
+
+        data = util.get_data_with_conversion(data, self.dtype)
+
+        cd = self.chunk_data
+        if not cd.has_data():
+            # nothing in file, so just append
+            fill_last_chunk = False
+        else:
+            # see if nrows is less than the chunksize
+            fill_last_chunk = (
+                cd['nrows'][-1] != self.row_chunksize
+            )
+
+        if fill_last_chunk:
+            # we need to append some to this chunk
+            num_to_append = self.row_chunksize - cd['nrows'][-1]
+
+            self._append_within_last_chunk(data[:num_to_append])
+
+            if num_to_append == data.size:
+                # we already appended all the data
+                return
+
+            adata = data[num_to_append:]
+        else:
+            adata = data
+
+        self._append_in_chunks(adata)
+
+    def vacuum(self):
+        """
+        Defragment the compressed data
+
+        When updating data in compressed columns, the new compressed chunks can
+        expand beyond their allocated region in the file.  In this case the new
+        compressed data is stored temporarily in a separate file.  Running
+        vacuum combines all data back together in a single contiguous file.
+        """
+        from tempfile import TemporaryDirectory
+        import shutil
+
+        cd = self.chunk_data
+        w, = np.where(cd['is_external'])
+        if w.size == 0:
+            return
+
+        with TemporaryDirectory(dir=self.dir) as tmpdir:
+
+            array_file = os.path.join(
+                tmpdir,
+                os.path.basename(self.filename),
+            )
+            chunks_file = os.path.join(
+                tmpdir,
+                os.path.basename(self.chunks_filename),
+            )
+            with open(array_file, 'wb') as fobj:
+                with ChunkData(chunks_file, mode='w+') as chunks:
+
+                    for ichunk in range(cd.size):
+
+                        compressed_bytes = self._read_compressed_bytes(ichunk)
+
+                        fobj.write(compressed_bytes)
+
+                        chunks.update_after_write(
+                            nrows=cd['nrows'][ichunk],
+                            nbytes=len(compressed_bytes),
+                        )
+            # close old files
+            self.chunk_data.close()
+            self._fobj.close()
+
+            # remove old files
+            print(f'    rm {self.filename}')
+            os.remove(self.filename)
+            print(f'    rm {self.chunks_filename}')
+            os.remove(self.chunks_filename)
+
+            for ei in w:
+                ext_fname = self.get_external_filename(ei)
+                print(f'    rm {ext_fname}')
+                os.remove(ext_fname)
+
+            # move new files into place
+            print(f'    {array_file} -> {self.filename}')
+            shutil.move(array_file, self.filename)
+            print(f'    {chunks_file} -> {self.chunks_filename}')
+            shutil.move(chunks_file, self.chunks_filename)
+
+        self._open_files()
+
+    def _append_in_chunks(self, data):
+        # Write data in chunks, the last one may be unfilled.
+        nwrites = data.size // self.row_chunksize
+        if data.size % self.row_chunksize != 0:
+            nwrites += 1
+
+        nrows_start = self.nrows
+        nrows_written = 0
+        for i in range(nwrites):
+            start = i * self.row_chunksize
+            end = (i + 1) * self.row_chunksize
+            data_to_write = data[start:end]
+            self._append(data_to_write)
+            nrows_written += data_to_write.size
+
+        nrec = self.nrows - nrows_start
+        if nrec != nrows_written:
+            raise RuntimeError(
+                f'wrote {nrows_written} but nrows incremented by {nrec}'
+            )
+
+    def _append_within_last_chunk(self, data):
+        """
+        append data within chunk. The new data must fit within a chunk, if the
+        input data overfills the chunk, an exception is raised
+        """
+
+        last_chunk_data = self.read_chunk(self.nchunks - 1)
+        new_chunk_data = np.hstack([last_chunk_data, data])
+
+        if new_chunk_data.size > self.row_chunksize:
+            raise ValueError(
+                f'appending last chunk got size {new_chunk_data.size} '
+                f'> chunksize {self.row_chunksize}'
+            )
+
+        # overwrite old chunk data and append ne
+        self._append(new_chunk_data, overwrite_last=True)
+
+        # the cache will be out of date
+        self._clear_chunk_cache()
+
+        if False:
+            import esutil as eu
+            tmp = self.read_chunk(self.nchunks - 1)
+            assert eu.numpy_util.compare_arrays(tmp, new_chunk_data)
+
+    def _append(self, data, overwrite_last=False):
+        """
+        Append data, either appending an entirely new chunk or overwriting the
+        last chunk when filling an incomplete one
+
+        Parameters
+        ----------
+        data: array
+            Data to be appended
+        overwrite_last: bool, optional
+            If set to True, start writing at the offset of the last chunk, to
+            overwrite existing data.  This is needed when updating a compressed
+            chunk.
+        """
+
+        cd = self.chunk_data
+
+        if self.nrows == 0:
+            offset = 0
+        else:
+            if overwrite_last:
+                # filling in an existing chunk
+                offset = cd['offset'][-1]
+            else:
+                # appending an entirely new chunk
+                offset = cd['offset'][-1] + cd['nbytes'][-1]
+
+        self._fobj.seek(offset)
+
+        if self.compression:
+            nbytes = self._write_compressed_data(data)
+        else:
+            nbytes = self._write_uncompressed_data(data)
+
+        assert self._fobj.tell() == offset + nbytes, (
+            'file tell == offset + nbytes'
+        )
+
+        self._update_chunks_after_write(
+            data.size, nbytes, overwrite_last=overwrite_last,
+        )
+
+    def _write_uncompressed_data(self, data):
+        data.tofile(self._fobj)
+        nbytes = data.nbytes
+        return nbytes
+
+    def _write_compressed_data(self, data, is_compressed=False):
+        if is_compressed:
+            compressed_bytes = data
+        else:
+            compressed_bytes = self._get_compressed_data(data)
+
+        self._fobj.write(compressed_bytes)
+        self._fobj.flush()
+        nbytes = len(compressed_bytes)
+        return nbytes
+
+    def _write_external_compressed_bytes(self, compressed_bytes, chunk_index):
+        fname = self.get_external_filename(chunk_index)
+
+        if True and self.verbose:
+            print(f'     writing external: {fname}')
+
+        with open(fname, 'wb') as fobj:
+            fobj.write(compressed_bytes)
+
+    def _read_external_compressed_bytes(self, chunk_index):
+        fname = self.get_external_filename(chunk_index)
+        with open(fname, 'rb') as fobj:
+            compressed_bytes = fobj.read()
+        return compressed_bytes
+
+    def get_external_filename(self, chunk_index):
+        """
+        External chunk index file
+        """
+        return f'{self.filename}.{chunk_index}'
+
+    def _get_compressed_data(self, data):
+        import blosc
+
+        if data.flags['C_CONTIGUOUS'] or data.flags['F_CONTIGUOUS']:
+            # this saves memory
+            compressed_bytes = blosc.compress_ptr(
+                data.__array_interface__['data'][0],
+                data.size,
+                data.dtype.itemsize,
+                **self.compression
+            )
+        else:
+            dbytes = data.tobytes()
+            compressed_bytes = blosc.compress(
+                dbytes,
+                data.dtype.itemsize,
+                **self.compression
+            )
+        return compressed_bytes
+
+    def _update_chunks_after_write(
+        self, nrows, nbytes, overwrite_last=False,
+    ):
+        """
+        Parameters
+        ----------
+        nrows: int
+            Number of rows now in most recent written chunk
+        nbytes: int
+            Number of bytes now in most recent written chunk
+        overwrite_last: bool
+            If set to True, we overwrote the last chunk so
+            also overwrite the chunk information
+        """
+
+        cd = self.chunk_data
+        if overwrite_last:
+            nrows_added = nrows - cd['nrows'][-1]
+        else:
+            nrows_added = nrows
+
+        self.chunk_data.update_after_write(
+            nrows=nrows, nbytes=nbytes, overwrite=overwrite_last,
+        )
+
+        self._check_file_position_after_append()
+
+        old_nrows = self.nrows
+        self._set_nrows()
+
+        assert self.nrows == old_nrows + nrows_added, (
+            'nrows == old_nrows + nrows'
+        )
+
+    def _check_file_position_after_append(self):
+        ftell = self._fobj.tell()
+
+        cd = self.chunk_data
+
+        predicted = cd['offset'][-1] + cd['nbytes'][-1]
+
+        if ftell != predicted:
+            s = repr(self)
+            raise RuntimeError(
+                f'predicted file position {predicted} but got {ftell} '
+                f'in column '
+                f'\n{s}'
+            )
+
+    def extend(self, nrows, fill=None):
+        """
+        Extend the column to the specified number of rows
+
+        Parameters
+        ----------
+        nrows: int
+            New number of rows
+        fill: fill value, optional
+            Optionally fill with this value
+        """
+        nrows_to_add = int(nrows) - self.nrows
+        if nrows_to_add <= 0:
+            return
+
+        # can choose this better
+        row_chunksize = self.row_chunksize
+        data = np.zeros(row_chunksize, dtype=self.dtype)
+        if fill is not None:
+            data[:] = fill
+
+        nwrites = nrows_to_add // row_chunksize
+        nleft = nrows_to_add % row_chunksize
+        if nleft > 0:
+            nwrites += 1
+
+        for i in range(nwrites):
+            if i == nwrites - 1:
+                self.append(data[:nleft])
+            else:
+                self.append(data)
+
+    def read_chunk(self, chunk_index):
+        """
+        Read the indicated chunk
+
+        Parameters
+        ----------
+        chunk_index: int
+            The index of the chunk
+
+        Returns
+        -------
+        array
+        """
+        # make a writeable copy
+        return self._read_chunk(chunk_index).copy()
+
+    def _read_chunk(self, chunk_index, writeable=False):
+        """
+        Read a chunk of data from the file
+
+        This version does not make a copy, but returns a readonly view.
+        """
+        self._cache_chunk(chunk_index)
+
+        if writeable:
+            # we need to clear the cache since its going to get updated
+            chunk = self._cached_chunk
+            self._clear_chunk_cache()
+            return chunk
+        else:
+            view = self._cached_chunk.view()
+            view.flags['WRITEABLE'] = False
+            return view
+
+    def _cache_chunk(self, chunk_index):
+        self._check_chunk(chunk_index)
+
+        if chunk_index == self._cached_chunk_index:
+            return
+
+        if self.compression:
+            chunk = self._read_compressed_chunk(chunk_index)
+        else:
+            chunk = self._read_uncompressed_chunk(chunk_index)
+
+        cd = self.chunk_data
+
+        expected = cd['nrows'][chunk_index]
+        if chunk.size != expected:
+            raise RuntimeError(
+                f'read {chunk.size} for chunk {chunk_index} '
+                f'but expected {expected}'
+            )
+        self._cached_chunk_index = chunk_index
+        self._cached_chunk = chunk
+
+    def _clear_chunk_cache(self):
+        self._cached_chunk_index = -1
+        self._cached_chunk = None
+
+    def _read_uncompressed_chunk(self, chunk_index):
+        cd = self.chunk_data
+
+        offset = cd['offset'][chunk_index]
+        nrows = cd['nrows'][chunk_index]
+        self._fobj.seek(offset, 0)
+        return np.fromfile(
+            self._fobj,
+            dtype=self.dtype,
+            count=nrows,
+        )
+
+    def _read_compressed_chunk(self, chunk_index):
+        import blosc
+
+        compressed_bytes = self._read_compressed_bytes(chunk_index)
+
+        cd = self.chunk_data
+        chunk = cd[chunk_index]
+
+        output = np.empty(chunk['nrows'], dtype=self.dtype)
+        blosc.decompress_ptr(
+            compressed_bytes, output.__array_interface__['data'][0]
+        )
+        return output
+
+    def _read_compressed_bytes(self, chunk_index):
+        cd = self.chunk_data
+
+        chunk = cd[chunk_index]
+
+        if chunk['is_external']:
+            buff = self._read_external_compressed_bytes(chunk_index)
+        else:
+            fobj = self._fobj
+            fobj.seek(chunk['offset'], 0)
+            nbytes = chunk['nbytes']
+
+            buff = bytearray(nbytes)
+
+            nread = fobj.readinto(buff)
+            if nread != len(buff):
+                raise RuntimeError(
+                    f'Expected to read {len(buff)} but read {nread}'
+                )
+
+        return buff
+
+    def _check_chunk(self, chunk_index):
+        if self.nrows == 0:
+            raise ValueError('file has no data')
+
+        if chunk_index > self.nchunks - 1:
+            raise ValueError(
+                f'chunk {chunk_index} out of range [0, {self.nchunks-1}')
+
+    def _set_compression(self, compression):
+        """
+        defaults get filled in
+        """
+        if not compression:
+            self._compression = False
+        else:
+            self._compression = util.get_compression_with_defaults(
+                compression,
+                convert=True,
+            )
+
+    def _set_row_chunksize(self, chunksize):
+        self._chunksize = chunksize
+        try:
+            end = chunksize[-1]
+            if end == 'r':
+                self._row_chunksize = int(chunksize[:-1])
+                return
+        except TypeError:
+            pass
+
+        chunksize_bytes = util.convert_to_bytes(chunksize)
+        chunksize_bytes = int(chunksize_bytes)
+
+        bytes_per_element = self.dtype.itemsize
+        self._row_chunksize = chunksize_bytes // (bytes_per_element * 2)
+        if self._row_chunksize < 1:
+            raise ValueError(
+                f'chunksize {chunksize} results in less than one row per chunk'
+            )
+
+    def _open_files(self):
+        self._chunk_data = ChunkData(
+            filename=self.chunks_filename,
+            mode=self.mode,
+        )
+
+        # needed to read/write bytes
+        mode = self.mode
+        if 'b' not in mode:
+            mode = 'b' + mode
+
+        self._fobj = open(self.filename, mode)
+
+    def _read_rows(self, data, rows):
+        cd = self.chunk_data
+
+        sortind = rows.sort_index
+        if sortind is not None:
+            # note original rows is not destroyed
+            rows = rows[sortind]
+
+        chunk_indices = util.get_chunks(
+            chunkrows_sorted=cd['rowstart'],
+            rows=rows,
+        )
+        # for this we assume rows are sorted
+        h, _ = np.histogram(chunk_indices, np.arange(self.nchunks+1))
+
+        w, = np.where(h > 0)
+        start = 0
+
+        for ci in w:
+            num = h[ci]
+            chunk = self._read_chunk(ci)
+
+            end = start + num
+            rowstart = cd['rowstart'][ci]
+
+            # this gives us rows within the chunk
+            trows = rows[start:end] - rowstart
+
+            if sortind is not None:
+                srows = sortind[start:end]
+                data[srows] = chunk[trows]
+            else:
+                data[start:end] = chunk[trows]
+
+            start += num
+
+        return data
+
+    def _set_rows(self, data, rows):
+        cd = self.chunk_data
+
+        if data.size == 1 and rows.size != 1:
+            fill_scalar = True
+        else:
+            fill_scalar = False
+
+        sortind = rows.sort_index
+        if sortind is not None:
+            # note original rows is not destroyed
+            rows = rows[sortind]
+
+        chunk_indices = util.get_chunks(
+            chunkrows_sorted=cd['rowstart'],
+            rows=rows,
+        )
+        # for this we assume rows are sorted
+        h, _ = np.histogram(chunk_indices, np.arange(self.nchunks+1))
+
+        w, = np.where(h > 0)
+        start = 0
+
+        for ci in w:
+            num = h[ci]
+
+            # settin writeable clears the cache
+            chunk = self._read_chunk(ci, writeable=True)
+
+            end = start + num
+            rowstart = cd['rowstart'][ci]
+
+            # this gives us rows within the chunk
+            trows = rows[start:end] - rowstart
+
+            if fill_scalar:
+                chunk[trows] = data
+            else:
+                if sortind is not None:
+                    srows = sortind[start:end]
+                    # data[srows] = chunk[trows]
+                    chunk[trows] = data[srows]
+                else:
+                    # data[start:end] = chunk[trows]
+                    chunk[trows] = data[start:end]
+
+            self._update_chunk(ci, chunk)
+
+            start += num
+
+    def _update_chunk(self, chunk_index, data):
+
+        cd = self.chunk_data
+
+        tnrows = cd['nrows'][chunk_index]
+        tnbytes = cd['nbytes'][chunk_index]
+
+        if data.size != tnrows:
+            raise ValueError(f'data size {data.size} != chunk nrows {tnrows}')
+
+        if not self.compression:
+            if data.nbytes != tnbytes:
+                raise ValueError(
+                    f'data nbytes {data.nbytes} != chunk nbytes {tnbytes}'
+                )
+
+            self._fobj.seek(cd['offset'][chunk_index])
+            nbytes = self._write_uncompressed_data(data)
+
+        else:
+            compressed_bytes = self._get_compressed_data(data)
+            nbytes = len(compressed_bytes)
+
+            is_ext = cd['is_external'][chunk_index]
+            needs_extending = chunk_index != cd.size-1 and nbytes > tnbytes
+
+            if is_ext or needs_extending:
+                self._write_external_compressed_bytes(
+                    compressed_bytes, chunk_index
+                )
+                cd.update_after_write(
+                    nrows=cd['nrows'][chunk_index],
+                    nbytes=None,  # not used due to is_external=True
+                    overwrite=True,
+                    chunk_index=chunk_index,
+                    is_external=True,
+                )
+            else:
+                # we can write directly into the chunk, just changing the
+                # nbytes entry
+                self._fobj.seek(cd['offset'][chunk_index])
+                self._write_compressed_data(
+                    compressed_bytes, is_compressed=True,
+                )
+                cd.update_after_write(
+                    nrows=tnrows,
+                    nbytes=nbytes,
+                    overwrite=True,
+                    chunk_index=chunk_index,
+                )
+
+    def __getitem__(self, arg):
+        from .indices import Indices
+
+        # returns either Indices or slice
+        # converts slice with step to indices
+        rows = util.extract_rows(arg, self.nrows)
+
+        # TODO, make an optimized slice reader
+        # for now, convert
+        if isinstance(rows, slice):
+            rows = Indices(np.arange(rows.start, rows.stop, rows.step))
+
+        data = np.empty(rows.size, dtype=self.dtype)
+
+        if rows.ndim == 0:
+            send_rows = Indices([rows])
+            self._read_rows(data, send_rows)
+            data = data[0]
+        else:
+            self._read_rows(data, rows)
+
+        return data
+
+    def __setitem__(self, arg, data):
+        """
+        Item lookup method, e.g. col[..] meaning slices or
+        sequences, etc.
+        """
+        from .indices import Indices
+
+        data = util.get_data_with_conversion(data, self.dtype)
+
+        # raise NotImplementedError(
+        #     'updating compressed columns not yet supported'
+        # )
+
+        # returns either Indices or slice
+        # converts slice with step to indices
+        rows = util.extract_rows(arg, self.nrows)
+
+        # TODO, make an optimized slice reader
+        # for now, convert
+        if isinstance(rows, slice):
+            rows = Indices(np.arange(rows.start, rows.stop, rows.step))
+
+        if rows.ndim == 0:
+            send_rows = Indices([rows])
+            self._set_rows(data, send_rows)
+            data = data[0]
+        else:
+            self._set_rows(data, rows)
+
+        return data
+
+    def _set_nrows(self):
+        cd = self.chunk_data
+
+        self._nrows = 0
+
+        if cd.has_data():
+            self._nrows = cd['rowstart'][-1] + cd['nrows'][-1]
+            assert self.nrows == cd['nrows'].sum(), (
+                'nrows == sum nrows'
+            )
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, exception_type, exception_value, traceback):
+        self.close()
+
+    def __repr__(self):
+        indent = '    '
+
+        rep = [f'filename: {self.filename}']
+        rep += [f'mode: {self.mode}']
+        rep += [f'dtype: {self.dtype.str}']
+        rep += [f'nrows: {self.nrows}']
+        rep += [f'chunksize: {self.chunksize}']
+        rep += [f'chunks_filename: {self.chunks_filename}']
+
+        rep = [indent + r for r in rep]
+
+        rep = ['Chunks:'] + rep
+        return '\n'.join(rep)
+
+
+class ChunkData(object):
+    """
+    Manage data about chunks in memory and in a file
+    """
+    def __init__(self, filename, mode='r'):
+        self._filename = filename
+        self._mode = mode
+        self._dtype = np.dtype(CHUNKS_DTYPE)
+
+        self._fobj = CColumn(
+            self.filename,
+            dtype=self.dtype,
+            mode=self.mode,
+        )
+
+        self._data = None
+        if self._fobj.nrows > 0:
+            self._data = self._fobj[:]
+
+    def update_after_write(
+        self,
+        nrows,
+        nbytes,
+        overwrite=False,
+        chunk_index=None,
+        is_external=False,
+    ):
+        """
+        Parameters
+        ----------
+        nrows: int
+            Number of rows now in most recent written chunk
+        nbytes: int
+            Number of bytes now in most recent written chunk
+        overwrite: bool
+            If set to True, we overwrote the chunk indicated
+            by chunk_index so also overwrite the chunk information
+        chunk_index: int, optional
+            Defaults to last
+        is_external: bool, optional
+            If set to True, indicates this chunk is stored externally
+            from the main file, happens when updates cause the chunk
+            to grow in compressed size.
+        """
+
+        if overwrite:
+            assert self._data is not None, (
+                'cannot overwrite last when starting first chunk'
+            )
+
+            if chunk_index is None:
+                chunk_index = self.size - 1
+
+            self['nrows'][chunk_index] = nrows
+
+            if is_external:
+                # we don't update nbytes, since it is now in its own file
+                # and we don't need to keep track of it
+                self['is_external'][chunk_index] = is_external
+            else:
+                self['nbytes'][chunk_index] = nbytes
+
+            self._fobj.update_row(chunk_index, self[chunk_index])
+        else:
+            chunk = np.zeros(1, dtype=self.dtype)
+
+            chunk['nbytes'] = nbytes
+            chunk['nrows'] = nrows
+
+            if self._data is None:
+                self._data = chunk
+            else:
+
+                chunk['offset'] = self['offset'][-1] + self['nbytes'][-1]
+                chunk['rowstart'] = self['rowstart'][-1] + self['nrows'][-1]
+
+                self._data = np.hstack([self._data, chunk])
+
+            self._fobj.append(chunk)
+
+    @property
+    def filename(self):
+        """
+        get the filename
+        """
+        return self._filename
+
+    @property
+    def mode(self):
+        """
+        get the file open mode
+        """
+        return self._mode
+
+    @property
+    def dtype(self):
+        """
+        get the numpy dtype of the chunks file data
+        """
+        return self._dtype
+
+    @property
+    def size(self):
+        """
+        get the numpy dtype of the chunks file data
+        """
+        self.ensure_has_data()
+        return self._data.size
+
+    def has_data(self):
+        """
+        Returns true if data is present
+        """
+        return self._data is not None
+
+    def ensure_has_data(self):
+        """
+        Raises exception if not data present
+        """
+        if not self.has_data():
+            raise RuntimeError('no chunk data exists')
+
+    def __getitem__(self, arg):
+        self.ensure_has_data()
+        return self._data[arg]
+
+    def close(self):
+        self._fobj.close()
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, exception_type, exception_value, traceback):
+        self.close()
+
+
+def test_set_compressed():
+    """
+    cache_mem of 0.01 will force use of mergesort
+    """
+    import tempfile
+    import numpy as np
+    from .columns import Columns
+    from .schema import TableSchema
+
+    seed = 333
+    # num = 20
+    # chunksize = '10r'
+    num = 100_000
+    chunksize = '10000r'
+
+    rng = np.random.RandomState(seed)
+
+    dtype = [('id', 'i8'), ('rand', 'f4'), ('scol', 'U5')]
+    data = np.zeros(num, dtype=dtype)
+    data['id'] = np.arange(num)
+    data['rand'] = rng.uniform(size=num)
+    data['scol'] = [
+        's' + str(data['id'][i]) for i in range(num)
+    ]
+
+    ccols = ['id', 'scol']
+
+    schema = TableSchema.from_array(
+        data, compression=ccols, chunksize=chunksize,
+    )
+
+    print(schema)
+
+    with tempfile.TemporaryDirectory() as tmpdir:
+
+        cdir = os.path.join(tmpdir, 'test.cols')
+        cols = Columns.create(cdir, schema, verbose=True)
+
+        # import IPython; IPython.embed()
+        cols.append(data)
+        assert np.all(cols['id'][:] == data['id'])
+
+        ndata = rng.randint(0, 2**16, size=cols.size)
+        cols['id'][:] = ndata
+        assert np.all(cols['id'][:] == ndata)
+
+        cols['id'][:] = 999
+        assert np.all(cols['id'][:] == 999)
+
+        print(f"nchunks: {cols['id']._col.nchunks}")
+        cols.append(data)
+        print(f"nchunks: {cols['id']._col.nchunks}")
+        assert np.all(cols['id'][data.size:] == data['id'])
+
+        ndata = rng.randint(0, 2**16, size=cols.size)
+        cols['id'][data.size:] = ndata
+        # import IPython; IPython.embed()
+
+        # print(cols['id']._col.chunk_data._data)
+        # stop
diff --git a/pycolumns/column.py b/pycolumns/column.py
index de7c051..42daec2 100644
--- a/pycolumns/column.py
+++ b/pycolumns/column.py
@@ -1,204 +1,26 @@
-"""
-TODO
-
-    - allow appending without updating index, which we would do at the end
-      of a series of appends
-
-"""
-import os
 import numpy as np
-
 from . import util
-from .indices import Indices
+from ._column import Column as CColumn
+from .chunks import Chunks
+from .defaults import DEFAULT_CACHE_MEM
 
 
-class ColumnBase(object):
-    """
-    Represent a column in a Columns database.  Facilitate opening, reading,
-    writing of data.  This class can be instantiated alone, but is usually
-    accessed through the Columns class.
-    """
-    def __init__(
-        self,
-        filename=None,
-        name=None,
-        dir=None,
-        verbose=False,
-    ):
-
-        self._do_init(
-            filename=filename,
-            name=name,
-            dir=dir,
-            verbose=verbose,
-        )
-
-    def _do_init(
-        self,
-        filename=None,
-        name=None,
-        dir=None,
-        verbose=False,
-    ):
-        """
-        See main docs for the Column class
-        """
-
-        self._clear()
-
-        self._filename = filename
-        self._name = name
-        self._dir = dir
-        self._verbose = verbose
-
-        # make sure we have something to work with before continuing
-        if not self._init_args_sufficient(filename, dir, name):
-            return
-
-        if self.filename is not None:
-            self._set_meta_from_filename()
-
-        elif self.name is not None:
-            self._set_meta_from_name()
-
-    @property
-    def verbose(self):
-        return self._verbose
-
-    @property
-    def dir(self):
-        """
-        get the directory holding the file
-        """
-        return self._dir
-
-    @property
-    def filename(self):
-        """
-        get the file name holding the column data
-        """
-        return self._filename
-
-    @property
-    def name(self):
-        """
-        get the name type of the column
-        """
-        return self._name
-
-    @property
-    def type(self):
-        """
-        get the data type of the column
-        """
-        return self._type
-
-    def reload(self):
-        """
-        Just reload the metadata for this column
-        """
-        if self.verbose:
-            print('Reloading column metadata for: %s' % self.name)
-        self._do_init(filename=self.filename, verbose=self.verbose)
-
-    def _clear(self):
-        """
-        Clear out all the metadata for this column.
-        """
-
-        self._filename = None
-        self._name = None
-        self._dir = None
-
-    def read(self, *args):
-        """
-        Read data from a column
-        """
-        raise NotImplementedError('implement read')
-
-    #
-    # setup methods
-    #
-
-    def _set_meta_from_filename(self):
-        """
-        Initiaize this column based on the pull path filename
-        """
-        if self.filename is not None:
-            if self.verbose:
-                print("Initializing from file: %s" % self.filename)
-
-            if self.filename is None:
-                raise ValueError("You haven't specified a filename yet")
-
-            self._name = util.extract_colname(self.filename)
-        else:
-            raise ValueError("You must set filename to use this function")
-
-    def _set_meta_from_name(self):
-        """
-        Initialize this based on name.  The filename is constructed from
-        the dir and name
-        """
-
-        if self.name is not None and self.dir is not None:
-
-            if self.verbose:
-                mess = "Initalizing from \n\tdir: %s \n\tname: %s \n\ttype: %s"
-                print(mess % (self.dir, self.name, self.type))
-
-            self._filename = util.create_filename(
-                self.dir,
-                self.name,
-                self.type,
-            )
-        else:
-            raise ValueError("You must set dir,name,type to use this function")
-
-    def _get_repr_list(self, full=False):
-        """
-
-        Get a list of metadat for this column.
-
-        """
-        raise NotImplementedError('implemente _get_repr_list')
-
-    def __repr__(self):
-        """
-        Print out some info about this column
-        """
-        s = self._get_repr_list(full=True)
-        s = "\n".join(s)
-        return s
-
-    def _init_args_sufficient(self,
-                              filename=None,
-                              dir=None,
-                              name=None):
-        """
-        Determine if the inputs are enough for initialization
-        """
-        if (filename is None) and \
-                (dir is None or name is None):
-            return False
-        else:
-            return True
-
-
-class ArrayColumn(ColumnBase):
+class Column(object):
     """
     Represent an array column in a Columns database
 
     Parameters
     ----------
-    filename: str, optional
-        Path to the file
-    name: str, optional
-        Name of the column
-    dir: str, optional
-        Directory of column
-    cache_mem: number, optional
-        Memory for cache used when creating index in gigabytes. Default 1.0
+    coldir: str
+        Path to the directory holding data
+    cache_mem: str or number
+        Cache memory for index creation, default '1g' or one gigabyte.
+        Can be a number in gigabytes or a string
+        Strings should be like '{amount}{unit}'
+            '1g' = 1 gigabytes
+            '100m' = 100 metabytes
+            '1000k' = 1000 kilobytes
+        Units can be g, m or k, case insenitive
     verbose: bool, optional
         If set to True print messages
 
@@ -207,10 +29,7 @@ class ArrayColumn(ColumnBase):
     # there are alternative construction methods
     # this method determines all info from the full path
 
-    col = ArrayColumn(filename='/full/path')
-
-    # this one uses directory, column name
-    col = ArrayColumn(name='something', dir='/path2o/dbname.cols')
+    col = Column('/full/path')
 
     Slice and item lookup
     ---------------------
@@ -241,137 +60,184 @@ class ArrayColumn(ColumnBase):
     """
     def __init__(
         self,
-        filename=None,
-        name=None,
-        dir=None,
-        cache_mem=1,
+        coldir,
+        cache_mem=DEFAULT_CACHE_MEM,
         verbose=False,
-    ):
-        self._do_init(
-            filename=filename,
-            name=name,
-            dir=dir,
-            cache_mem=cache_mem,
-            verbose=verbose,
-        )
-
-    def _do_init(
-        self,
-        filename=None,
-        name=None,
-        dir=None,
-        verbose=False,
-        cache_mem=1,
     ):
         """
         initialize the meta data, and possibly load the mmap
         """
 
-        self._type = 'array'
-        self._ext = 1
+        self._dir = coldir
+        self._cache_mem = cache_mem
+        self._cache_mem_gb = util.convert_to_gigabytes(cache_mem)
+        self._verbose = verbose
+        self._is_updating = False
+        self.reload()
 
-        self._cache_mem_gb = float(cache_mem)
+    def _close(self):
+        if hasattr(self, '_col') and self._col is not None:
+            self._col.close()
+            self._col = None
 
-        super()._do_init(
-            filename=filename,
-            name=name,
-            dir=dir,
-            verbose=verbose,
-        )
+    def reload(self):
+        """
+        load, or reload all meta data and reopen/open files
+        """
+
+        self._close()
+
+        path_info = util.get_colfiles(self.dir)
+
+        self._meta_filename = path_info['meta']
+        self._array_filename = path_info['array']
+        self._index_filename = path_info['index']
+        self._index1_filename = path_info['index1']
+        self._sorted_filename = path_info['sorted']
+        self._chunks_filename = path_info['chunks']
+        self._name = path_info['name']
 
-        # if self.filename is None or not os.path.exists(self.filename):
-        if self.filename is None:
-            return
+        if self.dir != path_info['dir']:
+            raise ValueError(
+                f'mismatch dir {dir} and path_info {path_info["dir"]}'
+            )
+
+        self._meta = util.read_json(self.meta_filename)
 
-        self._open_file()
+        self._type = 'col'
+        self._ext = 1
+        self._dtype = np.dtype(self._meta['dtype'])
+        self._index_dtype = np.dtype('i8')
+        self._index1_dtype = np.dtype([('index', 'i8'), ('value', self.dtype)])
+
+        self._open()
 
         # get info for index if it exists
         self._init_index()
 
-    def _open_file(self):
-        import fitsio
-        self._fits = fitsio.FITS(self.filename, 'rw')
-
-    def _get_hdu(self):
-        return self._fits[self._ext]
+    @property
+    def name(self):
+        """
+        get the name type of the column
+        """
+        return self._name
 
-    def _clear(self):
+    @property
+    def dir(self):
         """
-        Clear out all the metadata for this column.
+        get the directory holding the file
         """
+        return self._dir
 
-        super()._clear()
+    @property
+    def verbose(self):
+        return self._verbose
 
-        if self.has_data:
-            del self._fits
-            del self._dtype
+    @property
+    def is_updating(self):
+        return self._is_updating
 
-        self._has_index = False
+    @property
+    def type(self):
+        """
+        get the data type of the column
+        """
+        return self._type
 
     @property
-    def has_data(self):
+    def filenames(self):
         """
-        returns True if this column has some data
+        get a list of paths to all files associated with
+        this column
         """
-        has_data = False
+        fnames = [
+            self.meta_filename,
+            self.array_filename,
+        ]
 
-        if hasattr(self, '_fits'):
-            if len(self._fits) > 1:
-                hdu = self._get_hdu()
-                if hdu.has_data():
-                    has_data = True
-        # import IPython; IPython.embed()
-        return has_data
+        meta = self.meta
+        if 'compression' in meta and meta['compression']:
+            fnames += [self.chunks_filename]
 
-    def ensure_has_data(self):
+        if self.has_index:
+            fnames += [
+                self.index_filename,
+                self.index1_filename,
+                self.sorted_filename,
+            ]
+        return fnames
+
+    @property
+    def meta_filename(self):
         """
-        raise RuntimeError if no data is present
+        get the filename for the array of this column
         """
-        if not self.has_data:
-            raise ValueError('this column has no associated data')
+        return self._meta_filename
 
     @property
-    def dtype(self):
+    def array_filename(self):
         """
-        get the data type of the column
+        get the filename for the array of this column
         """
-        self.ensure_has_data()
+        return self._array_filename
 
-        if not hasattr(self, '_dtype'):
-            hdu = self._get_hdu()
-            dtype, _, _ = hdu.get_rec_dtype()
-            self._converted_dtype, self._convert_unicode = (
-                util.maybe_convert_ascii_dtype_to_unicode(dtype)
-            )
-            descr = dtype.descr
-            assert len(descr) == 1
-            assert descr[0][0] == 'data'
-            self._dtype = dtype
+    @property
+    def chunks_filename(self):
+        """
+        get the filename for the chunks for this column
+        """
+        return self._chunks_filename
 
-        return self._dtype
+    @property
+    def index_filename(self):
+        """
+        get the filename for the index of this column
+        """
+        return self._index_filename
 
     @property
-    def index_dtype(self):
-        return get_index_dtype(self.dtype)
+    def index1_filename(self):
+        """
+        get the filename for the hierarch 1 index of this column
+        """
+        return self._index1_filename
 
     @property
-    def nrows(self):
+    def sorted_filename(self):
         """
-        get the shape of the column
+        get the filename for the sorted version of this column
         """
-        self.ensure_has_data()
+        return self._sorted_filename
 
-        hdu = self._get_hdu()
-        return hdu.get_nrows()
+    @property
+    def meta(self):
+        """
+        get a copy of the meta data dict
+        """
+        return self._meta.copy()
 
     @property
-    def shape(self):
+    def dtype(self):
         """
-        get the shape of the column
+        get the data type of the column
         """
-        self.ensure_has_data()
+        return self._dtype
+        # return self._meta['dtype']
 
-        return (self.nrows, )
+    @property
+    def index_dtype(self):
+        return self._index_dtype
+
+    @property
+    def index1_dtype(self):
+        return self._index1_dtype
+
+    @property
+    def nrows(self):
+        """
+        get the number of rows in the column
+        """
+        return self._col.nrows
 
     @property
     def size(self):
@@ -382,7 +248,7 @@ def size(self):
 
     @property
     def data_size_bytes(self):
-        return self.dtype.itemsize * self.size
+        return self.dtype.itemsize * self.nrows
 
     @property
     def data_size_gb(self):
@@ -398,12 +264,80 @@ def index_size_gb(self):
 
     @property
     def cache_mem(self):
+        return self._cache_mem
+
+    @property
+    def cache_mem_gb(self):
         return self._cache_mem_gb
 
-    def _append(self, data):
+    def _open(self):
+        meta = self._meta
+        if 'compression' in meta and meta['compression']:
+            self._col = Chunks(
+                filename=self.array_filename,
+                chunks_filename=self.chunks_filename,
+                dtype=self._meta['dtype'],
+                mode='r+',
+                compression=meta['compression'],
+                chunksize=meta['chunksize'],
+                verbose=self.verbose,
+            )
+        else:
+            self._col = CColumn(
+                self.array_filename,
+                dtype=self._meta['dtype'],
+                mode='r+',
+                # verbose=self.verbose,
+            )
+
+    def resize(self, nrows):
         """
-        Append data to the column.  Data are appended if the file already
-        exists.
+        Expand or truncate the file to num rows.  If expanded, the
+        new rows are filled with the default value for this column.
+
+        The fill value is by default zero (0 for numbers, '' for strings)
+        but can be set in the schema as 'fill_value'
+
+        Parameters
+        ----------
+        nrows: int
+            The new number of rows.
+        """
+        if isinstance(self._col, CColumn):
+            nrows_old = self.nrows
+
+            # if extended, fills with zeros
+            self._col.resize(nrows)
+
+            # can fill with user defined value
+            if nrows > nrows_old and 'fill_value' in self._meta:
+                self[nrows_old:] = self._meta['fill_value']
+        else:
+            if nrows < self.nrows:
+                raise NotImplementedError('cannot shrink compressed columns')
+
+            self._col.extend(nrows, fill=self._meta.get('fill_value'))
+
+        if self.has_index:
+            self.update_index()
+
+    def vacuum(self):
+        """
+        Defragment compressed data
+
+        When updating data in compressed columns, the new compressed chunks can
+        expand beyond their allocated region in the file.  In this case the new
+        compressed data is stored temporarily in a separate file.  Running
+        vacuum combines all data back together in a single contiguous file.
+        """
+        if isinstance(self._col, Chunks):
+            self._col.vacuum()
+
+    def _append(self, data, update_index=True):
+        """
+        Append data to the column.  Data are appended.  If the file has not
+        been initialized, the data type is initialized to that of the input
+        data.
 
         Parameters
         ----------
@@ -412,29 +346,44 @@ def _append(self, data):
             the data types must match exactly.
         """
 
-        # make sure the data type of the input equals that of the column
-        if not isinstance(data, np.ndarray):
-            raise ValueError("For 'array' columns data must be a numpy array")
+        self._check_data(data)
 
-        if data.dtype.names is not None:
-            raise ValueError('do not enter data with fields')
+        # we use data conversion in self._col
+        # self._check_data_dtype(data)
+        self._col.append(data)
 
-        if data.ndim > 1:
-            raise ValueError('data must be one dimensional array')
+        if update_index and not self.is_updating:
+            self.update_index()
 
-        self._write_data_as_rec(data)
+    def updating(self, vacuum=False):
+        """
+        This enteres the updating context, which delays index
+        updates until after exiting the context
 
-        if self.has_index:
-            self._update_index()
+        with col.updating():
+            col[5:10] = 25
+            col[35:88] = 99
+        """
+        self._is_updating = True
+        self._vacuum_on_exit = vacuum
+        return self
 
-    def _write_data_as_rec(self, data):
-        view_data = self._get_rec_view(data)
+    def _check_data(self, data):
+        if data.dtype.names is not None:
+            raise ValueError('column data cannot have fields')
 
-        if not self.has_data:
-            self._fits.write(view_data)
-        else:
-            hdu = self._fits[self._ext]
-            hdu.append(view_data)
+        if not isinstance(data, np.ndarray):
+            raise ValueError(f'data must be a numpy array, got {type(data)}')
+
+        if data.ndim > 1:
+            raise ValueError('data must be one dimensional array')
+
+    # def _check_data_dtype(self, data):
+    #     dt = data.dtype
+    #     mydt = self.dtype
+    #
+    #     if dt != mydt:
+    #         raise ValueError(f"data dtype '{dt}' != '{mydt}'")
 
     def _get_rec_view(self, data):
         view_dtype = [('data', data.dtype.descr[0][1])]
@@ -445,37 +394,19 @@ def __getitem__(self, arg):
         Item lookup method, e.g. col[..] meaning slices or
         sequences, etc.
         """
-
-        self.ensure_has_data()
-
-        hdu = self._get_hdu()
-
-        rows = util.extract_rows(arg, sort=True)
-
-        if rows is None:
-            data = hdu[:]['data']
-        elif isinstance(rows, slice):
-            data = hdu[rows]['data']
-        else:
-            data = np.zeros(rows.size, dtype=self.dtype)
-            hdu._FITS.read_rows_as_rec(self._ext+1, data, rows)
-            if self._convert_unicode:
-                data = data.astype(self._converted_dtype, copy=False)
-            data = data['data']
-            if rows.ndim == 0:
-                data = data[0]
-
-        return data
+        return self._col[arg]
 
     def __setitem__(self, arg, values):
         """
         Item lookup method, e.g. col[..] meaning slices or
         sequences, etc.
         """
-        raise RuntimeError('fix writing')
+        self._col[arg] = values
 
-        if not hasattr(self, '_fits'):
-            raise ValueError('no file loaded yet')
+        # todo, context manager for this so only updates index after leaving
+        # context
+        if not self.is_updating:
+            self.update_index()
 
     def read(self, rows=None):
         """
@@ -486,18 +417,15 @@ def read(self, rows=None):
         rows: sequence, slice, Indices or None, optional
             A subset of the rows to read.
         """
-        if not hasattr(self, '_fits'):
-            raise ValueError('no file loaded yet')
-
-        return self[rows]
+        return self._col[rows]
 
     def _delete(self):
         """
         Attempt to delete the data file associated with this column
         """
 
-        if hasattr(self, '_sf'):
-            del self._fits
+        if hasattr(self, '_col'):
+            del self._col
 
         super()._delete()
 
@@ -522,74 +450,91 @@ def verify_index_available(self):
         if not self.has_index:
             raise ValueError('no index exists for column %s' % self.name)
 
-    def create_index(self):
+    def create_index(self, overwrite=False):
         """
         Create an index for this column.
 
-        Once the index is created, all data from the column will be put into
-        the index.  Also, after creation any data appended to the column are
-        automatically added to the index.
+        Parameters
+        ----------
+        overwrite: bool, optional
+            If set to True, overwrite any existing index.  See also
+            update_index()
         """
-        import tempfile
+        import os
+        from tempfile import TemporaryDirectory
         import shutil
 
-        if self.has_index:
-            print('column %s already has an index')
-            return
+        if self.has_index and not overwrite:
+            raise RuntimeError(
+                f'column {self.name} already has an index.  Send '
+                f' overwrite=True or use update_index()'
+            )
 
-        if len(self.shape) > 1:
-            raise ValueError('cannot index a multi-dimensional column')
+        self.delete_index()
+
+        # total usage for in-memory sorting is index size plus twice the size
+        # data since we need to do data[s] to get the sorted this is an
+        # optimization to not run sort twice
+        size_gb = self.index_size_gb + self.data_size_gb * 2
 
         if self.verbose:
-            print('creating index for column %s' % self.name)
-            print('index size gb:', self.index_size_gb)
-            print('cache mem gb:', self._cache_mem_gb)
+            print('creating index for column', self.name)
+            print(f'cache mem: {self.cache_mem}')
+            print(f'required gb: {size_gb:.3g}')
 
-        with tempfile.TemporaryDirectory(dir=self.dir) as tmpdir:
-            tfile = os.path.join(
+        with TemporaryDirectory(dir=self.dir) as tmpdir:
+            ifile = os.path.join(
                 tmpdir,
                 os.path.basename(self.index_filename),
             )
-            if self.index_size_gb < self._cache_mem_gb:
-                self._write_index_memory(tfile)
-            else:
-                self._write_index_mergesort(tmpdir, tfile)
+            sfile = os.path.join(
+                tmpdir,
+                os.path.basename(self.sorted_filename),
+            )
+            with CColumn(ifile, mode='w+', dtype=self.index_dtype) as ifobj:
+                with CColumn(sfile, mode='w+', dtype=self.dtype) as sfobj:
 
-            if self.verbose:
-                print(f'{tfile} -> {self.index_filename}')
+                    if size_gb < self.cache_mem_gb:
+                        self._write_index_memory(ifobj, sfobj)
+                    else:
+                        self._write_index_mergesort(tmpdir, ifobj, sfobj)
 
-            shutil.move(tfile, self.index_filename)
+                    if self.verbose:
+                        print(f'  {ifile} -> {self.index_filename}')
+                    shutil.move(ifile, self.index_filename)
 
-        self._init_index()
+                    if self.verbose:
+                        print(f'  {sfile} -> {self.sorted_filename}')
+                    shutil.move(sfile, self.sorted_filename)
 
-    def _write_index_memory(self, fname):
-        import fitsio
+        self._write_index1()
+        self._init_index()
 
+    def _write_index_memory(self, ifobj, sfobj):
         if self.verbose:
             print(f'creating index for {self.name} in memory')
 
-        # want native so no copying which happens for writes
-        # dt = self.index_dtype
-        dt = get_index_dtype(self.dtype, native=True)
-        index_data = np.zeros(self.shape[0], dtype=dt)
-        index_data['index'] = np.arange(index_data.size)
-        index_data['value'] = self[:]
+        data = self[:]
+        if self.verbose:
+            print('  sorting')
+
+        sort_index = data.argsort()
 
-        index_data.sort(order='value')
+        if self.verbose:
+            print('  writing')
 
-        # set up file name and data type info
-        with fitsio.FITS(fname, mode='rw', clobber=True) as output:
-            output.write(index_data)
+        ifobj.append(sort_index)
+        sfobj.append(data[sort_index])
 
-    def _write_index_mergesort(self, tmpdir, fname):
+    def _write_index_mergesort(self, tmpdir, ifobj, sfobj):
         from .mergesort import create_mergesort_index
 
         if self.verbose:
             print(f'creating index for {self.name} with mergesort on disk')
 
-        chunksize_bytes = int(self._cache_mem_gb * 1024**3)
+        chunksize_bytes = int(self.cache_mem_gb * 1024**3)
 
-        bytes_per_element = self.index_dtype.itemsize
+        bytes_per_element = self.index_dtype.itemsize + self.dtype.itemsize
 
         # need factor of two because we keep both the cache and the scratch in
         # mergesort
@@ -597,28 +542,67 @@ def _write_index_mergesort(self, tmpdir, fname):
 
         create_mergesort_index(
             source=self,
-            outfile=fname,
+            isink=ifobj,
+            ssink=sfobj,
             chunksize=chunksize,
             tmpdir=tmpdir,
             verbose=self.verbose,
         )
 
+    def _write_index1(self, chunksize_rows=10_000):
+        """
+        this needs to be worked out much better
+        """
+        nchunks = self.nrows // chunksize_rows
+        if self.nrows % chunksize_rows != 0:
+            nchunks += 1
+
+        if nchunks < 10:
+            rows = np.arange(self.nrows)
+        else:
+            rows = np.ones(nchunks+1, dtype='i8')
+            rows[0:-1] = np.arange(0, self.nrows, chunksize_rows)
+            rows[-1] = self.nrows - 1
+
+        output = np.zeros(rows.size, dtype=self.index1_dtype)
+        output['index'] = rows
+
+        if self.verbose:
+            print('  reading index1 values from:', self.sorted_filename)
+        with CColumn(self.sorted_filename, dtype=self.dtype) as scol:
+            output['value'] = scol[rows]
+
+        if self.verbose:
+            print('  writing:', self.index1_filename)
+        with CColumn(
+                self.index1_filename,
+                mode='w+',
+                dtype=self.index1_dtype) as i1col:
+            i1col.append(output)
+
     def update_index(self):
         """
-        re-create the index
+        Recreate the index for this column.
         """
-        self.delete_index()
-        self.create_index()
+        if self.has_index:
+            self.create_index(overwrite=True)
 
     def delete_index(self):
         """
         Delete the index for this column if it exists
         """
+        import os
+
         if self.has_index:
-            index_fname = self.index_filename
-            if os.path.exists(index_fname):
-                print("Removing index for column: %s" % self.name)
-                os.remove(index_fname)
+            names = [
+                self.index_filename,
+                self.index1_filename,
+                self.sorted_filename,
+            ]
+            for name in names:
+                if os.path.exists(name):
+                    print("Removing: %s" % name)
+                    os.remove(name)
 
         self._init_index()
 
@@ -626,29 +610,41 @@ def _init_index(self):
         """
         If index file exists, load some info
         """
-        import fitsio
+        import os
+        import numpy as np
+
+        self._arr1 = np.zeros(1, dtype=self.dtype)
+
+        if os.path.exists(self.index_filename):
+            if not os.path.exists(self.sorted_filename):
+                raise RuntimeError(
+                    f'missing sorted file {self.sorted_filename}'
+                )
+
+            if not os.path.exists(self.index_filename):
+                raise RuntimeError(
+                    f'missing index file {self.index_filename}'
+                )
+
+            if not os.path.exists(self.index1_filename):
+                raise RuntimeError(
+                    f'missing index1 file {self.index1_filename}'
+                )
 
-        index_fname = self.index_filename
-        if os.path.exists(index_fname):
             self._has_index = True
-            self._index = fitsio.FITS(index_fname, 'rw')
-            self._iarr1 = np.zeros(1, dtype=self.index_dtype)
+            self._index = CColumn(self.index_filename, dtype=self.index_dtype)
+            self._index1 = CColumn(
+                self.index1_filename, dtype=self.index1_dtype,
+            )
+            self._sorted = CColumn(self.sorted_filename, dtype=self.dtype)
+
+            self._index1_data = self._index1[:]
         else:
             self._has_index = False
             self._index = None
-
-    @property
-    def index_filename(self):
-        """
-        get the filename for the index of this column
-        """
-        if self.filename is None:
-            return None
-
-        # remove the final extension
-        index_fname = '.'.join(self.filename.split('.')[0:-1])
-        index_fname = index_fname+'.index'
-        return index_fname
+            self._index1 = None
+            self._index1_data = None
+            self._sorted = None
 
     def match(self, values):
         """
@@ -657,15 +653,17 @@ def match(self, values):
         Parameters
         ----------
         values: scalar or array
-            Value or values to match.  These entries should be
-            unique to avoid unexpected results.
+            Value or values to match.
 
         Returns
         -------
         Indices of matches
         """
+        import numpy as np
+        from .indices import Indices
 
-        values = np.array(values, ndmin=1, copy=False)
+        # this makes an array
+        values = np.unique(values)
 
         # query each separately
         ind_list = []
@@ -698,41 +696,67 @@ def __eq__(self, val):
         return self.between(val, val)
 
     def _read_one_from_index(self, index):
-        iarr1 = self._iarr1
-        self._index._FITS.read_as_rec(self._ext+1, index+1, index+1, iarr1)
-        val = iarr1['value'][0]
-        if self._convert_unicode:
-            return str(val, 'utf-8')
+        arr1 = self._arr1
+        self._sorted._read_row(arr1, index)
+        val = arr1[0]
         return val
 
     def _bisect_right(self, val):
-        return _bisect_right(
-            func=self._read_one_from_index,
-            x=val,
-            lo=0,
-            hi=self.nrows,
-        )
+        """
+        bisect on the hierarch index1 then possibly the full index
+        """
+        import bisect
+        idata = self._index1_data
+        i1 = bisect.bisect_right(idata['value'], x=val)
+
+        if i1 == 0 or i1 == idata.size:
+            sind = i1
+        else:
+            start = idata['index'][i1 - 1]
+            end = idata['index'][i1]
+
+            sind = _bisect_right_func(
+                func=self._read_one_from_index,
+                x=val,
+                lo=start,
+                hi=end,
+            )
+
+        return sind
 
     def _bisect_left(self, val):
-        return _bisect_left(
-            func=self._read_one_from_index,
-            x=val,
-            lo=0,
-            hi=self.nrows,
-        )
+        """
+        bisect on the hierarch index1 then possibly the full index
+        """
+        import bisect
+        idata = self._index1_data
+        i1 = bisect.bisect_left(idata['value'], x=val)
+
+        if i1 == 0 or i1 == idata.size:
+            sind = i1
+        else:
+            start = idata['index'][i1 - 1]
+            end = idata['index'][i1]
+
+            sind = _bisect_left_func(
+                func=self._read_one_from_index,
+                x=val,
+                lo=start,
+                hi=end,
+            )
+
+        return sind
 
     # one-sided range operators
     def __gt__(self, val):
         """
         bisect_right returns i such that data[i:] are all strictly > val
         """
+        from .indices import Indices
+
         self.verify_index_available()
         i = self._bisect_right(val)
-        indices = self._index[1]['index'][i:].copy()
-
-        # mmap = self._index.mmap
-        # i = bisect.bisect_right(mmap['value'], val)
-        # indices = mmap['index'][i:].copy()
+        indices = self._index[i:]
 
         return Indices(indices)
 
@@ -740,14 +764,12 @@ def __ge__(self, val):
         """
         bisect_left returns i such that data[i:] are all strictly >= val
         """
+        from .indices import Indices
+
         self.verify_index_available()
 
         i = self._bisect_left(val)
-        indices = self._index[1]['index'][i:].copy()
-
-        # mmap = self._index.mmap
-        # i = bisect.bisect_left(mmap['value'], val)
-        # indices = mmap['index'][i:].copy()
+        indices = self._index[i:]
 
         return Indices(indices)
 
@@ -755,14 +777,12 @@ def __lt__(self, val):
         """
         bisect_left returns i such that data[:i] are all strictly < val
         """
+        from .indices import Indices
+
         self.verify_index_available()
 
         i = self._bisect_left(val)
-        indices = self._index[1]['index'][:i].copy()
-
-        # mmap = self._index.mmap
-        # i = bisect.bisect_left(mmap['value'], val)
-        # indices = mmap['index'][:i].copy()
+        indices = self._index[:i]
 
         return Indices(indices)
 
@@ -770,14 +790,12 @@ def __le__(self, val):
         """
         bisect_right returns i such that data[:i] are all strictly <= val
         """
+        from .indices import Indices
+
         self.verify_index_available()
 
         i = self._bisect_right(val)
-        indices = self._index[1]['index'][:i].copy()
-
-        # mmap = self._index.mmap
-        # i = bisect.bisect_right(mmap['value'], val)
-        # indices = mmap['index'][:i].copy()
+        indices = self._index[:i]
 
         return Indices(indices)
 
@@ -818,186 +836,86 @@ def between(self, low, high, interval='[]'):
         # index array
         ind = columns.where( (col1.between(low,high)) & (col2 == value2) )
         """
+        from .indices import Indices
 
         self.verify_index_available()
 
-        # mmap = self._index.mmap
         if interval == '[]':
             # bisect_left returns i such that data[i:] are all strictly >= val
-            # ilow = bisect.bisect_left(mmap['value'], low)
             ilow = self._bisect_left(low)
 
             # bisect_right returns i such that data[:i] are all strictly <= val
-            # ihigh = bisect.bisect_right(mmap['value'], high)
             ihigh = self._bisect_right(high)
 
         elif interval == '(]':
             # bisect_right returns i such that data[i:] are all strictly > val
-            # ilow = bisect.bisect_right(mmap['value'], low)
             ilow = self._bisect_right(low)
 
             # bisect_right returns i such that data[:i] are all strictly <= val
-            # ihigh = bisect.bisect_right(mmap['value'], high)
             ihigh = self._bisect_right(high)
 
         elif interval == '[)':
             # bisect_left returns i such that data[:i] are all strictly >= val
-            # ilow = bisect.bisect_left(mmap['value'], low)
             ilow = self._bisect_left(low)
 
             # bisect_left returns i such that data[:i] are all strictly < val
-            # ihigh = bisect.bisect_left(mmap['value'], high)
             ihigh = self._bisect_left(high)
 
         elif interval == '()':
             # bisect_right returns i such that data[i:] are all strictly > val
-            # ilow = bisect.bisect_right(mmap['value'], low)
             ilow = self._bisect_right(low)
 
             # bisect_left returns i such that data[:i] are all strictly < val
-            # ihigh = bisect.bisect_left(mmap['value'], high)
             ihigh = self._bisect_left(high)
         else:
             raise ValueError('bad interval type: %s' % interval)
 
-        # indices = mmap['index'][ilow:ihigh].copy()
-        indices = self._index[1]['index'][ilow:ihigh]
+        indices = self._index[ilow:ihigh]
 
         return Indices(indices)
 
-    def _get_repr_list(self, full=False):
-        """
+    def __enter__(self):
+        self._is_updating = True
+        return self
 
-        Get a list of metadat for this column.
+    def __exit__(self, exception_type, exception_value, traceback):
+        self._is_updating = False
+        self.update_index()
+        if self._vacuum_on_exit:
+            self.vacuum()
 
-        """
+    def __repr__(self):
         indent = '  '
 
-        if not full:
-            s = ''
-            if self.name is not None:
-                s += 'Column: %-15s' % self.name
-
-            s += ' type: %10s' % self.type
-            if self.has_data:
-
-                # if self.shape is not None:
-                #     s += ' shape: %12s' % (self.shape,)
-                s += ' nrows: %12s' % self.nrows
+        s = []
+        if self.name is not None:
+            s += ['name: %s' % self.name]
 
-                if self.name is not None:
-                    s += ' has index: %s' % self.has_index
-
-            s = [s]
-        else:
-            s = []
-            if self.name is not None:
-                s += ['name: %s' % self.name]
+        s += ['filename: %s' % self.array_filename]
+        s += ['type: array']
+        s += ['index: %s' % self.has_index]
 
-            if self.filename is not None:
-                s += ['filename: %s' % self.filename]
+        if 'compression' in self.meta and self.meta['compression']:
+            c = self.meta['compression']
+            s += ['compression:']
+            s += ['    cname: %s' % c['cname']]
+            s += ['    clevel: %s' % c['clevel']]
+            s += ['    shuffle: %s' % c['shuffle']]
+            s += ['chunksize: %s' % self._col.chunksize]
 
-            s += ['type: array']
+        c_dtype = self.dtype.descr[0][1]
+        s += ['dtype: %s' % c_dtype]
 
-            if self.has_data:
-                s += ['has index: %s' % self.has_index]
+        s += ['nrows: %s' % self.nrows]
 
-                c_dtype = self.dtype.descr[0][1]
-                s += ['dtype: %s' % c_dtype]
-
-                # if self.shape is not None:
-                #     s += ['shape: %s' % (self.shape,)]
-                s += ['nrows: %s' % self.nrows]
-
-            s = [indent + tmp for tmp in s]
-            s = ['Column: '] + s
-
-        return s
-
-
-class DictColumn(ColumnBase):
-    def _do_init(
-        self,
-        filename=None,
-        name=None,
-        dir=None,
-        verbose=False,
-    ):
-
-        self._type = 'dict'
-
-        super()._do_init(
-            filename=filename,
-            name=name,
-            dir=dir,
-            verbose=verbose,
-        )
-
-    def write(self, data):
-        """
-        Write data to the dict column.
-
-        Parameters
-        ----------
-        data: dict or json supported object
-            The data must be supported by the JSON format.
-        """
-
-        util.write_json(data, self.filename)
-
-    def read(self):
-        """
-        read data from this column
-        """
-
-        return util.read_json(self.filename)
-
-    def _get_repr_list(self, full=False):
-        """
-
-        Get a list of metadat for this column.
-
-        """
-        indent = '  '
-
-        if not full:
-            s = ''
-            if self.name is not None:
-                s += 'Column: %-15s' % self.name
-            s += ' type: %10s' % self.type
-
-            s = [s]
-        else:
-            s = []
-            if self.name is not None:
-                s += ['name: %s' % self.name]
-
-            if self.filename is not None:
-                s += ['filename: %s' % self.filename]
-
-            s += ['type: dict']
-
-            s = [indent + tmp for tmp in s]
-            s = ['Column: '] + s
+        s = [indent + tmp for tmp in s]
+        s = ['Column: '] + s
 
+        s = "\n".join(s)
         return s
 
 
-def get_index_dtype(dtype, native=False):
-    # this removed the <,  > etc so it
-    # is native
-    dt = dtype.descr[0][1]
-    if native:
-        dt = dt[1:]
-
-    return np.dtype([
-        ('index', 'i8'),
-        # ('value', dtype.descr[0][1][1:]),
-        ('value', dt),
-    ])
-
-
-def _bisect_right(func, x, lo, hi):
+def _bisect_right_func(func, x, lo, hi):
     """
     bisect right with function call to get value
     """
@@ -1012,7 +930,7 @@ def _bisect_right(func, x, lo, hi):
     return lo
 
 
-def _bisect_left(func, x, lo, hi):
+def _bisect_left_func(func, x, lo, hi):
     """
     bisect left with function call to get value
     """
diff --git a/pycolumns/columns.py b/pycolumns/columns.py
index f0fba79..7d6cf8e 100644
--- a/pycolumns/columns.py
+++ b/pycolumns/columns.py
@@ -1,23 +1,10 @@
-"""
-todo
-
-    - allow update index without completely redoing it
-    - tests for sub columns
-    - Maybe add option "unsort" to put indices back in original unsorted order
-    - add updating a set of columns/column
-    - make it possible to add_column for array if we then eventually verify
-    - if reading single row, scalar, doing from column gives a number but
-      on columns with read gives length 1 array
-    - support update on dict column, which would be like a normal dict
-      update
-"""
 import os
-from glob import glob
 import numpy as np
-from .column import ColumnBase, ArrayColumn, DictColumn
+from .column import Column
+from .metafile import Meta
 from . import util
-
-ALLOWED_COL_TYPES = ['array', 'dict', 'cols']
+from .schema import TableSchema
+from .defaults import DEFAULT_CACHE_MEM, DEFAULT_CHUNKSIZE
 
 
 class Columns(dict):
@@ -26,21 +13,286 @@ class Columns(dict):
 
     Parameters
     ----------
-    dir: str
-        Path to database
-    cache_mem: number, optional
-        Cache memory for index creation in gigabytes.  Default 1.0
+    coldir: str
+        Path to the Columns store.  The directory must already exist.
+        Use Columns.create to create a new Columns store
+    cache_mem: str or number
+        Cache memory for index creation, default '1g' or one gigabyte.
+        Can be a number in gigabytes or a string
+        Strings should be like '{amount}{unit}'
+            '1g' = 1 gigabytes
+            '100m' = 100 metabytes
+            '1000k' = 1000 kilobytes
+        Units can be g, m, k or b case insenitive
     verbose: bool, optional
         If set to True, print messages
     """
 
-    def __init__(self, dir=None, cache_mem=1, verbose=False):
+    def __init__(self, coldir, cache_mem=DEFAULT_CACHE_MEM, verbose=False):
+
+        coldir = os.path.expandvars(coldir)
+
+        if not os.path.exists(coldir):
+            raise RuntimeError(
+                f'Directory {coldir} does not exist.  Use Columns.create to '
+                f'initialize'
+            )
+
+        self._dir = coldir
         self._type = 'cols'
         self._verbose = verbose
-        self._cache_mem_gb = float(cache_mem)
-        self._set_dir(dir)
+        self._is_updating = False
+        self._cache_mem = cache_mem
         self._load()
 
+    @classmethod
+    def create(
+        cls, coldir, cache_mem=DEFAULT_CACHE_MEM, verbose=False,
+        yes=False,
+    ):
+        """
+        Initialize a new columns database.  The new Columns object
+        is returned.
+
+        Parameters
+        ----------
+        coldir: str
+            Path to columns directory
+        cache_mem: str or number
+            Cache memory for index creation, default '1g' or one gigabyte.
+            Can be a number in gigabytes or a string
+            Strings should be like '{amount}{unit}'
+                '1g' = 1 gigabytes
+                '100m' = 100 metabytes
+                '1000k' = 1000 kilobytes
+            Units can be g, m, k or b case insenitive
+
+        verbose: bool, optional
+            If set to True, display information
+        yes: bool, optional
+            If set to True, do not prompt for confirmation when overwriting
+            an existing directory
+
+        Examples
+        --------
+        import pycolumns as pyc
+
+        cols = pyc.Columns.create(coldir)
+        """
+        import shutil
+
+        coldir = os.path.expandvars(coldir)
+
+        if os.path.exists(coldir):
+            if not yes:
+                answer = input(
+                    f'directory {coldir} already exists, are you sure '
+                    f'you want to overwrite? (y/n)'
+                )
+                if answer.lower() == 'y':
+                    yes = True
+
+            if not yes:
+                return
+
+            if verbose:
+                print(f'removing {coldir}')
+
+            shutil.rmtree(coldir)
+
+        if verbose:
+            print(f'creating: {coldir}')
+
+        os.makedirs(coldir)
+
+        cols = Columns(coldir, cache_mem=cache_mem, verbose=verbose)
+        return cols
+
+    @classmethod
+    def create_from_array(
+        cls,
+        coldir,
+        data,
+        name=None,
+        compression=None,
+        chunksize=DEFAULT_CHUNKSIZE,
+        append=True,
+        cache_mem=DEFAULT_CACHE_MEM,
+        verbose=False,
+        yes=False,
+
+    ):
+        """
+        Initialize a new columns database and create a table from the
+        input data.
+
+        Parameters
+        ----------
+        coldir: str
+            Path to columns directory
+        data: numpy array with fields or dict of arrays
+            The data from which to derive the schema
+        name: str, optional
+            Name for the new Columns with trailin slash, e.g. sub/ or
+            sub1/sub2/.  If not sent, will be created in the root directory
+        compression: dict, list, bool, or None, optional
+            See TableSchema.from_array for a full explanation
+        chunksize: dict, str or number
+            See TableSchema.from_array for a full explanation
+        append: bool, optional
+            If set to True, the data are also written to the new
+            Default True
+
+        cache_mem: str or number
+            Cache memory for index creation, default '1g' or one gigabyte.
+            Can be a number in gigabytes or a string
+            Strings should be like '{amount}{unit}'
+                '1g' = 1 gigabytes
+                '100m' = 100 metabytes
+                '1000k' = 1000 kilobytes
+            Units can be g, m, k or b case insenitive
+        verbose: bool, optional
+            If set to True, display information
+        yes: bool, optional
+            If set to True, do not prompt for confirmation when overwriting
+            an existing directory
+
+        Examples
+        --------
+        array = np.zeros(num, dtype=[('id', 'i8'), ('x', 'f4')])
+        cols = Columns.create_from_array(dir, array)
+        """
+
+        cols = Columns.create(
+            coldir=coldir,
+            cache_mem=cache_mem,
+            verbose=verbose,
+            yes=yes,
+        )
+
+        cols.from_array(
+            data=data,
+            name=name,
+            compression=compression,
+            chunksize=chunksize,
+            append=append,
+            yes=yes,
+        )
+        return cols
+
+    def create_table(
+        self,
+        schema={},
+        name=None,
+        yes=False,
+    ):
+        """
+        Initialize a new columns database table.  The new Columns object
+        is returned.
+
+        Parameters
+        ----------
+        schema: dict, optional
+            Dictionary holding information for each column.  If not sent,
+            or empty, no columns are created
+        name: str, optional
+            Name for the new Columns, e.g. 'subcols/' or 'subcols1/subcols2/'.
+            If not sent, will be created in the root directory of the
+            Columns.  If not None, the name must end in a slash '/'
+        yes: bool, optional
+            If set to True, do not prompt for confirmation when overwriting
+            an existing directory
+
+        Examples
+        --------
+        import pycolumns as pyc
+
+        # create from column schema
+        idcol = pyc.ColumnSchema(dtype='i8', compress=True)
+        racol = pyc.ColumnSchema(dtype='f8')
+        schema = pyc.TableSchema([idcol, racol])
+        cols = pyc.Columns.create_table(schema=schema)
+
+        # give it a path
+        cols = pyc.Columns.create_table(schema=schema, name='subcols/')
+
+        # see the TableSchema and ColumnSchema classes for more options
+        """
+
+        subdir = util.get_sub_dir(self.dir, name)
+
+        if subdir is not None:
+            cols = Columns.create(
+                subdir,
+                cache_mem=self.cache_mem,
+                verbose=self.verbose,
+                yes=yes,
+            )
+        else:
+            cols = self
+
+        cols._add_columns(schema)
+
+        if subdir is not None:
+            # Need a full reload because a number of levels can be affected
+            self._load()
+
+        return cols
+
+    def from_array(
+        self,
+        data,
+        name=None,
+        compression=None,
+        chunksize=DEFAULT_CHUNKSIZE,
+        append=True,
+        yes=False,
+    ):
+        """
+        Initialize a new columns database, creating the schema from the input
+        array with fields.  The data from array are also written unless
+        append=False.
+
+        The new Columns object is returned.
+
+        Parameters
+        ----------
+        data: numpy array with fields or dict of arrays
+            The data from which to derive the schema
+        name: str, optional
+            Name for the new Columns with trailin slash, e.g. sub/ or
+            sub1/sub2/.  If not sent, will be created in the root directory
+        compression: dict, list, bool, or None, optional
+            See TableSchema.from_array for a full explanation
+        chunksize: dict, str or number
+            See TableSchema.from_array for a full explanation
+        append: bool, optional
+            If set to True, the data are also written to the new
+            Default True
+
+        yes: bool, optional
+            If set to True, do not prompt for confirmation when overwriting
+            an existing directory
+
+        Examples
+        --------
+        cols = Columns(coldir)
+        array = np.zeros(num, dtype=[('id', 'i8'), ('x', 'f4')])
+        cols.from_array(array, name='data/observations/')
+        """
+        schema = TableSchema.from_array(
+            data, compression=compression, chunksize=chunksize,
+        )
+        cols = self.create_table(
+            schema=schema,
+            name=name,
+            yes=yes,
+        )
+        if append:
+            cols.append(data)
+
+        return cols
+
     @property
     def nrows(self):
         """
@@ -48,6 +300,13 @@ def nrows(self):
         """
         return self._nrows
 
+    @property
+    def size(self):
+        """
+        number of rows in table
+        """
+        return self.nrows
+
     @property
     def names(self):
         """
@@ -60,17 +319,17 @@ def column_names(self):
         """
         Get a list of all column names
         """
-        return [c for c in self if self[c].type == 'array']
+        return [c for c in self if self[c].type == 'col']
 
     @property
-    def dict_names(self):
+    def meta_names(self):
         """
         Get a list of the array column names
         """
-        return [c for c in self if self[c].type == 'dict']
+        return list(self.meta.keys())
 
     @property
-    def subcols_names(self):
+    def sub_table_names(self):
         """
         Get a list of the array column names
         """
@@ -92,82 +351,72 @@ def verbose(self):
         return self._verbose
 
     @property
-    def cache_mem(self):
-        return self._cache_mem_gb
+    def is_updating(self):
+        return self._is_updating
 
-    def _set_dir(self, dir=None):
-        """
-        Set the database directory, creating if none exists.
-        """
-        if dir is not None:
-            dir = os.path.expandvars(dir)
-
-        self._dir = dir
-
-        if not os.path.exists(dir):
-            os.makedirs(dir)
+    @property
+    def cache_mem(self):
+        return self._cache_mem
 
-    def delete(self, yes=False):
+    @property
+    def meta(self):
         """
-        delete the entire Columns database
-
-        Parameters
-        ----------
-        yes: bool
-            If True, don't prompt for confirmation
+        Do this properly to add protections and controls
         """
-        if not yes:
-            answer = input('really delete all data? (y/n) ')
-            if answer.lower() == 'y':
-                yes = True
-
-        if not yes:
-            return
-
-        for name in self:
-            self.delete_column(name, yes=True)
+        return self._meta
 
     def _dirbase(self):
         """
         Return the dir basename minus any extension
         """
-        if self.dir is not None:
-            bname = os.path.basename(self.dir)
-            name = '.'.join(bname.split('.')[0:-1])
-            return name
-        else:
-            return None
+        bname = os.path.basename(self.dir)
+        name = '.'.join(bname.split('.')[0:-1])
+        return name
 
-    def _load(self):
+    def _load(self, verify=True):
         """
-
-        Load all existing columns in the
-        directory.  Column files must have the right extensions to be noticed
-        so far these can be
-
-            .col
-            .json
-
-        and other column directories can be loaded if they have the extension
-
-            .cols
-
+        Load all entries
         """
-
         # clear out the existing columns and start from scratch
-        self._clear()
-
-        for type in ALLOWED_COL_TYPES:
-            if self.dir is not None:
-                pattern = os.path.join(self.dir, '*.'+type)
-                fnames = glob(pattern)
-
-                for f in fnames:
-                    if type == 'cols':
-                        self._load_coldir(f)
-                    else:
-                        self._load_column(f)
-        self.verify()
+        self._clear_all()
+
+        paths = os.listdir(self.dir)
+
+        # load the table columns
+        for i, path in enumerate(paths):
+            path = os.path.join(self.dir, path)
+            c = None
+            if util.is_column(path):
+                name = os.path.basename(path)
+                c = Column(
+                    path, cache_mem=self.cache_mem, verbose=self.verbose,
+                )
+                super().__setitem__(name, c)
+            else:
+
+                # will be /name for name.cols
+                # else will be name
+                name, ext = util.split_ext(path)
+                # name = util.extract_name(path)
+                # ext = util.extract_extension(path)
+
+                if self.verbose and ext in ['cols', 'json']:
+                    print(f'    loading : {name}')
+
+                # only load .dict or .cols, ignore other stuff
+                if ext == 'json':
+                    self.meta._load(path)
+                elif ext == 'cols':
+                    c = Columns(
+                        path, cache_mem=self.cache_mem, verbose=self.verbose,
+                    )
+                    cname = util.get_sub_name(name)
+                    super().__setitem__(cname, c)
+                else:
+                    pass
+
+        if verify:
+            self.verify()
 
     def verify(self):
         """
@@ -175,9 +424,11 @@ def verify(self):
         """
 
         first = True
+        self._nrows = 0
+
         for c in self.keys():
             col = self[c]
-            if col.type == 'array':
+            if col.type == 'col':
                 this_nrows = col.nrows
                 if first:
                     self._nrows = this_nrows
@@ -189,109 +440,45 @@ def verify(self):
                             'got %d vs %d' % (c, this_nrows, self.nrows)
                         )
 
-    def _load_column(self, filename):
+    def create_column(self, schema):
         """
-        Load the specified column
-        """
-
-        name = util.extract_colname(filename)
-        type = util.extract_coltype(filename)
-
-        if type not in ALLOWED_COL_TYPES:
-            raise ValueError("bad column type: '%s'" % type)
-
-        col = self._open_column(
-            filename=filename,
-            name=name,
-            type=type,
-        )
-
-        name = col.name
-        self._clear(name)
-        self[name] = col
+        Create a new column, filling with zeros or the fill value to the right
+        number of rows
 
-    def _open_column(self, filename, name, type):
-        if type == 'array':
-            col = ArrayColumn(
-                filename=filename,
-                dir=self.dir,
-                name=name,
-                verbose=self.verbose,
-                cache_mem=self.cache_mem,
-            )
-        elif type == 'dict':
-            col = DictColumn(
-                filename=filename,
-                dir=self.dir,
-                name=name,
-                verbose=self.verbose,
-            )
-        else:
-            raise ValueError("bad column type '%s'" % type)
-
-        return col
+        Parameters
+        ----------
+        schema: ColumnSchema
+            A ColumnSchema object
 
-    def create_column(self, name, type, verify=True):
+            schema = ColumnSchema(name='x', dtype='f4')
+            cols.create_column(schema)
         """
-        create the specified column
+        table_schema = TableSchema([schema])
+        self._add_columns(table_schema, fill=True)
 
-        You usually don't want to use this directly for array types if columns
-        already exist, because consistency will be broken, at least
-        temporarily. In fact an exception will be raised.  Better to use the
-        append method to ensure row length consistency
-
-        It is fine to use it for dict types
+    def create_meta(self, name, data={}):
+        """
+        create a new metadata entry
 
         parameters
         ----------
         name: str
-            Column name
-        type: str
-            Column type, 'dict'
+            Name for this metadata entry
+        data: Data that can be stored as JSON
+            e.g. a dict, list etc.
         """
 
-        if name in self:
+        if name in self.meta_names:
             raise ValueError("column '%s' already exists" % name)
 
-        if type not in ALLOWED_COL_TYPES:
-            raise ValueError("bad column type: '%s'" % type)
-
-        if self.dir is None:
-            raise ValueError("no dir is set for Columns db, can't "
-                             "construct names")
-
-        filename = util.create_filename(self.dir, name, type)
-
-        col = self._open_column(
-            filename=filename,
-            name=name,
-            type=type,
-        )
-
-        name = col.name
-        self[name] = col
-
-        if verify:
-            self.verify()
-
-    def _load_coldir(self, dir):
-        """
-        Load a coldir under this coldir
-        """
-        if not os.path.exists(dir):
-            raise RuntimeError("coldir does not exists: '%s'" % dir)
-
-        cols = Columns(
-            dir, cache_mem=self.cache_mem, verbose=self.verbose,
-        )
-
-        name = cols._dirbase()
-        self._clear(name)
-        self[name] = cols
+        path = util.get_filename(dir=self.dir, name=name, ext='json')
+        self.meta._load(path)
+        self.meta[name].write(data)
 
     def reload(self, columns=None):
         """
-        reload the database or a subset of columns
+        reload the database or a subset of columns.  Note discovery
+        of new entries is not performed
 
         parameters
         ----------
@@ -307,31 +494,95 @@ def reload(self, columns=None):
         for column in columns:
             if column not in self:
                 raise ValueError("Unknown column '%s'" % column)
+
             self[column].reload()
 
         self.verify()
 
-    def _clear(self, name=None):
+    def clear(self):
+        raise RuntimeError('clear() not supported on Columns')
+
+    def _clear_all(self, name=None):
         """
         Clear out the dictionary of column info
         """
-        if name is not None:
-            if isinstance(name, (list, tuple)):
-                for n in name:
-                    if n in self:
-                        del self[n]
-        else:
-            if name in self:
-                del self[name]
+        self._close()
+        super().clear()
+        self._meta = _MetaSet(self.dir, verbose=self.verbose)
+
+    def _close(self):
+        """
+        close all open files, including those of sub Columns directories
+        """
+        for col in self.column_names:
+            self[col]._close()
+
+        for scol in self.sub_table_names:
+            self[scol]._close()
+
+    def _add_columns(self, schema, fill=False):
+        """
+        Initialize new columns.  Currently this must be done while all other
+        columns are zero size to have consistency
+
+        Parameters
+        ----------
+        schema: TableSchema or dict
+            Schema holding information for each column.
+        fill: bool, optional
+            If set to True, the new columns are filled out to the current nrows
+        """
+
+        # Try to convert to schema
+        if not isinstance(schema, TableSchema):
+            schema = TableSchema.from_schema(schema)
+
+        for name, this in schema.items():
+            if self.verbose:
+                print('    creating:', name)
+
+            coldir = util.get_column_dir(self.dir, name)
+            if not os.path.exists(coldir):
+                os.makedirs(coldir)
+
+            metafile = util.get_filename(coldir, name, 'meta')
+            if os.path.exists(metafile):
+                raise RuntimeError(f'column {name} already exists')
+
+            util.write_json(metafile, this)
+
+            dfile = util.get_filename(coldir, name, 'array')
+            with open(dfile, 'w') as fobj:  # noqa
+                # just to create the empty file
+                pass
+
+            if 'compression' in this:
+                cfile = util.get_filename(coldir, name, 'chunks')
+                with open(cfile, 'w') as fobj:  # noqa
+                    # just to create the empty file
+                    pass
+
+        if fill and self.nrows > 0:
+            self._load(verify=False)
+            for name in schema:
+                self[name].resize(self.nrows)
+
+        self._load()
 
     def append(self, data, verify=True):
         """
-        Append data for the fields of a structured array to columns
+        Append data to the table columns
+
+        Parameters
+        ----------
+        data: array or dict
+            A structured array or dict with arrays
+        verify: bool, optional
+            If set to True, verify all the columns have the same number of rows
+            after appending.  Default True
         """
 
-        names = data.dtype.names
-        if names is None:
-            raise ValueError('append() takes a structured array as input')
+        names = util.get_data_names(data)
 
         if len(self) > 0:
             # make sure the input data matches the existing column names
@@ -339,7 +590,7 @@ def append(self, data, verify=True):
             column_names = set(self.column_names)
             if in_names != column_names:
                 raise ValueError(
-                    f'input columns {in_names}'
+                    f'input columns {in_names} '
                     f'do not match existing table columns {column_names}'
                 )
 
@@ -357,97 +608,141 @@ def _append_column(self, name, data):
         """
 
         if name not in self:
-            self.create_column(name, 'array', verify=False)
+            raise RuntimeError(f'column {name} does not exist')
 
-        self[name]._append(data)
+        self[name]._append(data, update_index=not self.is_updating)
 
-    def from_fits(self, filename, ext=1, lower=False):
+    def updating(self):
         """
-        Write columns to the database, reading from the input fits file.
-        Uses chunks of 100MB
+        This enteres the updating context, which delays index
+        updates until after exiting the context
 
-        parameters
-        ----------
-        filename: string
-            Name of the file to read
-        ext: extension number, optional
-            The FITS extension to read from
-        lower: bool, optional
-            if True, lower-case all names
+        with cols.updating():
+            cols.append(data1)
+            cols.append(data2)
         """
-        import fitsio
+        self._is_updating = True
+        return self
 
-        with fitsio.FITS(filename, lower=lower) as fits:
-            hdu = fits[ext]
+    def vacuum(self):
+        """
+        Degragment compressed columns
 
-            one = hdu[0:0+1]
-            nrows = hdu.get_nrows()
-            rowsize = one.itemsize
+        When updating data in compressed columns, the new compressed chunks can
+        expand beyond their allocated region in the file.  In this case the new
+        compressed data is stored temporarily in a separate file.  Running
+        vacuum combines all data back together in a single contiguous file.
+        """
+        for name in self.column_names:
+            self[name].vacuum()
 
-            # step size in bytes
-            step_bytes = int(self._cache_mem_gb * 1024**3)
+    def delete(self, yes=False):
+        """
+        delete the entire Columns database
 
-            # step size in rows
-            step = step_bytes // rowsize
+        Parameters
+        ----------
+        yes: bool
+            If set to True, don't prompt for confirmation
+        """
+        import shutil
 
-            nstep = nrows // step
-            nleft = nrows % step
+        if not yes:
+            answer = input('really delete all data? (y/n) ')
+            if answer.lower() == 'y':
+                yes = True
 
-            if nleft > 0:
-                nstep += 1
+        if not yes:
+            return
 
-            if self.verbose:
-                print('Loading %s rows from file: %s' % (nrows, filename))
+        original_names = list(self.keys())
+        for name in original_names:
+            self.delete_entry(name, yes=True)
 
-            for i in range(nstep):
+        for name in self.meta:
+            self.delete_meta(name, yes=True)
 
-                start = i * step
-                stop = (i + 1) * step
+        print('removing:', self.dir)
+        shutil.rmtree(self.dir)
 
-                if stop > nrows:
-                    # not needed, but use for printouts
-                    stop = nrows
+    def delete_meta(self, name, yes=False):
+        """
+        delete the specified dict
 
-                if self.verbose:
-                    print(f'    {start}:{stop} of {nrows}')
+        parameters
+        ----------
+        name: string
+            Name of entry to delete
+        yes: bool
+            If True, don't prompt for confirmation
+        """
 
-                data = hdu[start:stop]
+        if name not in self.meta:
+            print("cannot delete dict '%s', it does not exist" % name)
 
-                # not needed with fits backend
-                # data = util.get_native_data(data)
+        if not yes:
+            answer = input("really delete dict '%s'? (y/n) " % name)
+            if answer.lower() == 'y':
+                yes = True
 
-                self.append(data, verify=False)
+        if yes:
+            print(f'Removing data for dict: {name}')
+            fname = self.meta[name].filename
+            if os.path.exists(fname):
+                os.remove(fname)
 
-        self.verify()
+            del self.meta[name]
 
-    def delete_column(self, name, yes=False):
+    def delete_entry(self, name, yes=False):
         """
-        delete the specified column and reload
+        delete the specified entry and reload
 
         parameters
         ----------
         name: string
-            Name of column to delete
+            Name of entry to delete
         yes: bool
             If True, don't prompt for confirmation
         """
-        if name not in self.colnumn_names:
-            print("cannot delete column '%s', it does not exist" % name)
+        import shutil
+
+        if name not in self.names:
+            print("cannot delete entry '%s', it does not exist" % name)
 
         if not yes:
-            answer = input("really delete column '%s'? (y/n) " % name)
+            answer = input("really delete entry '%s'? (y/n) " % name)
             if answer.lower() == 'y':
                 yes = True
 
         if not yes:
             return
 
-        fname = self[name].filename
-        if os.path.exists(fname):
-            print("Removing data for column: %s" % name)
-            os.remove(fname)
+        entry = self[name]
+        entry._close()
 
-        self.reload()
+        if entry.type == 'cols':
+            print(f'Removing data for sub columns: {name}')
+
+            # have the sub cols remove data in case the subcols dir is a sym
+            # link
+            dname = entry.dir
+
+            entry.delete(yes=True)
+            if os.path.exists(dname):
+                shutil.rmtree(dname)
+
+        else:
+            # remove individual files in case it the directory is a sym link
+            print(f'Removing data for entry: {name}')
+            for fname in entry.filenames:
+                if os.path.exists(fname):
+                    print(f'    Removing: {fname}')
+                    os.remove(fname)
+
+            print(f'Removing: {entry.dir}')
+            shutil.rmtree(entry.dir)
+
+        del self[name]
 
     def read(
         self,
@@ -462,20 +757,22 @@ def read(
         ----------
         columns: sequence or string
             Can be a scalar string or a sequence of strings.  Defaults to all
-            array columns if asdict is False, but will include
-            dicts if asdict is True
+            array columns.
         rows: sequence, slice or scalar
             Sequence of row numbers.  Defaults to all.
         asdict: bool, optional
-            If set to True, read the requested columns into a dict.
+            If set to True, read the requested columns into a dict rather
+            than structured array
         """
 
-        columns = self._extract_columns(columns=columns, asdict=asdict)
+        columns = self._extract_columns(columns=columns)
 
         if len(columns) == 0:
             return None
 
-        rows = util.extract_rows(rows=rows, sort=True)
+        # converts to slice or Indices and converts stepped slices to
+        # arange
+        rows = util.extract_rows(rows=rows, nrows=self.nrows)
 
         if asdict:
             # Just putting the arrays into a dictionary.
@@ -488,27 +785,17 @@ def read(
 
                 # just read the data and put in dict, simpler than below
                 col = self[colname]
-                if col.type == 'array':
+                if col.type == 'col':
                     data[colname] = col.read(rows=rows)
                 else:
                     data[colname] = col.read()
 
         else:
 
-            if rows is not None:
-                if isinstance(rows, slice):
-                    n_rows2read = rows.stop - rows.start
-                else:
-                    try:
-                        n_rows2read = rows.size
-                    except TypeError:
-                        n_rows2read = 1
+            if isinstance(rows, slice):
+                n_rows2read = rows.stop - rows.start
             else:
-                for c in columns:
-                    col = self[c]
-                    if col.type == 'array':
-                        n_rows2read = col.nrows
-                        break
+                n_rows2read = rows.size
 
             # copying into a single array with fields
             dtype = self._extract_dtype(columns)
@@ -524,185 +811,269 @@ def read(
 
         return data
 
+    def list(self, full=False, isroot=True, indent=''):
+        """
+        List the directory structure
+
+        Parameters
+        ----------
+        full: bool
+            If set to True, list everything including column names
+            Metadata entries are enclosed in {brackets}
+        """
+        if isroot and not full:
+            # print('root')
+            nc = len(self.column_names)
+            nm = len(self.meta)
+            if nc > 0 or nm > 0:
+                ln = ['root has']
+                if nc > 0:
+                    ln += [f'{nc} columns']
+                if nm > 0:
+                    ln += [f'{nm} metadata']
+                ln = ' '.join(ln)
+                print(ln)
+
+        if full:
+            for name in self.column_names:
+                print(indent + '- ' + name)
+            for name in self.meta:
+                print(indent + '- ' + '{%s}' % name)
+
+        for name in self.names:
+            if name[-1] == '/':
+                print(indent + name)
+                self[name].list(isroot=False, full=full, indent=indent + '  ')
+
     def _extract_dtype(self, columns):
         dtype = []
 
         for colname in columns:
             col = self[colname]
-            shape = col.shape
 
-            descr = col.dtype.descr[0][1]
+            descr = col.dtype.str
 
             dt = (colname, descr)
-            if len(shape) > 1:
-                dt = dt + shape[1:]
-
             dtype.append(dt)
 
-        dtype, _ = util.maybe_convert_ascii_dtype_to_unicode(np.dtype(dtype))
         return dtype
 
-    read_columns = read
-
-    def read_column(self, colname, rows=None):
-        """
-        Only numpy, fixed length for now.  Eventually allow pickled columns.
-        """
-        if colname not in self:
-            raise ValueError("Column '%s' not found" % colname)
-
-        return self[colname].read(rows=rows)
-
-    def _extract_columns(self, columns=None, asdict=False):
+    def _extract_columns(self, columns=None):
         """
-        extract the columns to read.  If no columns are sent then the behavior
-        depends on the asdict parameter
-            - if asdict is False, read all array columns
-            - if asdict is True, read all columns
+        extract the columns to read.  If no columns are sent then all
+        columns are read
         """
         if columns is None:
-
-            if asdict:
-                keys = sorted(self.keys())
-                columns = keys
-            else:
-                columns = self.column_names
-
+            return self.column_names
         else:
             if isinstance(columns, str):
                 columns = [columns]
 
             for c in columns:
-                if c not in self:
+                if c not in self.column_names:
                     raise ValueError("Column '%s' not found" % c)
 
-                if not asdict and self[c].type != 'array':
-                    if not asdict:
-                        raise ValueError(
-                            "requested non-array column '%s.' "
-                            "Use asdict=True to read non-array "
-                            "columns with general read() method" % c
-                        )
-
         return columns
 
-    def _get_repr_list(self):
+    def __contains__(self, name):
+        try:
+            _ = self[name]
+            return True
+        except IndexError:
+            return False
+
+    def __getitem__(self, arg):
+
+        if not np.isscalar(arg):
+            if util.iscols(arg):
+                return _ColumnSubset(self, arg)
+            else:
+                return self.read(rows=arg)
+        else:
+            name = arg
+
+        notfound = False
+
+        if name in ['.', './']:
+            return self
+
+        # take off leading ./
+        if name not in self.names and len(name) > 2 and name[:2] == './':
+            name = name[2:]
+
+        if name not in self.names:
+            ns = name.split('/')
+            if len(ns) == 1:
+                # not a sub-columns, so not found
+                notfound = True
+            elif len(ns) == 2 and ns[1] == '':
+                # it was something like 'sub/'
+                notfound = True
+            else:
+                # It is multi-level
+                first = f'{ns[0]}/'
+                rest = '/'.join(ns[1:])
+
+                # let it pass on down the chain
+                try:
+                    return self[first][rest]
+                except IndexError:
+                    notfound = True
+        if notfound:
+            raise IndexError(f'entry {name} not found')
+
+        return super().__getitem__(name)
+
+    def __setitem__(self, name, data):
+        """
+        Only supported for dict.  For Column you need to do
+        cols[name][ind] = 3 etc.
+        """
+        item = self[name]
+        if item.type == 'dict':
+            item.write(data)
+        elif item.type == 'col':
+            # let the error handling occur in Column
+            self[name][:] = data
+        elif item.type == 'cols':
+            raise TypeError(
+                f'Attempt to replace entire sub Columns "{name}". '
+                f'If you are trying to set items for a Column inside '
+                f'{name}, use cols[{name}][indices] = data etc.'
+            )
+        else:
+            raise TypeError('Columns object does not support item assignment')
+
+    def __enter__(self):
+        self._is_updating = True
+        return self
+
+    def __exit__(self, exception_type, exception_value, traceback):
+        self._is_updating = False
+
+        for name in self.column_names:
+            self[name].update_index()
+
+    def __repr__(self):
         """
         Get a list of metadata for this columns directory and it's
         columns
         """
+        ncols = len(self.column_names)
         indent = '  '
         s = []
         if self.dir is not None:
-            dbase = self._dirbase()
-            s += [dbase]
+            # dbase = self._dirbase()
+            # s += [dbase]
             s += ['dir: '+self.dir]
-            if hasattr(self, '_nrows'):
+            if ncols > 0 and hasattr(self, '_nrows'):
                 s += ['nrows: %s' % self.nrows]
 
-        s += ['']
-        subcols = []
-        if len(self) > 0:
-            s += ['Columns:']
-            cnames = 'name', 'dtype', 'index'
-            s += ['  %-15s %6s %-6s' % cnames]
-            s += ['  '+'-'*(28)]
+        metas = []
+        if len(self.meta_names) > 0:
+            metas += ['Metadata:']
+            metas += ['  %-15s' % ('name',)]
+            metas += ['  '+'-'*(28)]
+            metas += ['  '+n for n in self.meta_names]
 
-            dicts = ['Dictionaries:']
-            dicts += ['  %-15s' % ('name',)]
-            dicts += ['  '+'-'*(28)]
-
-            subcols = ['Sub-Columns Directories:']
+        subcols = []
+        if len(self.sub_table_names) > 0:
+            subcols += ['Sub Tables:']
             subcols += ['  %-15s' % ('name',)]
             subcols += ['  '+'-'*(28)]
+            subcols += ['  '+n for n in self.sub_table_names]
+
+        acols = []
+
+        column_names = self.column_names
+        if len(column_names) > 0:
+            acols += ['Table Columns:']
+            cnames = 'name', 'dtype', 'comp', 'index'
+            acols += ['  %-15s %6s %7s %-6s' % cnames]
+            acols += ['  '+'-'*(35)]
 
-            for name in sorted(self):
+            for name in sorted(column_names):
                 c = self[name]
-                if isinstance(c, ColumnBase):
-
-                    name = c.name
-
-                    if len(name) > 15:
-                        name_entry = ['  %s' % name]
-                        # s += ['%23s' % (c.type,)]
-                    else:
-                        name_entry = ['  %-15s' % c.name]
-
-                    if c.type == 'array':
-                        s += name_entry
-                        c_dtype = c.dtype.descr[0][1]
-                        s[-1] += ' %6s' % c_dtype
-                        s[-1] += ' %-6s' % self[name].has_index
-                        # s[-1] += ' %s' % self[name].nrows
-                    elif c.type == 'dict':
-                        dicts += name_entry
-                    else:
-                        raise ValueError(f'bad type: {c.type}')
+
+                name = c.name
+
+                if len(name) > 15:
+                    # name_entry = ['  %s' % name]
+                    name_entry = [f'  {name}\n' + ' '*19]
+                    # s += ['%23s' % (c.type,)]
                 else:
-                    cdir = os.path.basename(c.dir).replace('.cols', '')
-                    subcols += ['  %s' % cdir]
+                    name_entry = ['  %-15s' % c.name]
+
+                if 'compression' in c.meta:
+                    comp = c.meta['compression']['cname']
+                else:
+                    comp = 'None'
+
+                acols += name_entry
+                c_dtype = c.dtype.descr[0][1]
+                acols[-1] += ' %6s' % c_dtype
+                acols[-1] += ' %7s' % comp
+                acols[-1] += ' %-6s' % self[name].has_index
 
         s = [indent + tmp for tmp in s]
-        s = ['Columns Directory: '] + s
+        s = ['Columns: '] + s
 
-        if len(dicts) > 3:
+        if len(acols) > 3:
             s += [indent]
-            dicts = [indent + tmp for tmp in dicts]
-            s += dicts
+            acols = [indent + tmp for tmp in acols]
+            s += acols
 
-        if len(subcols) > 3:
+        if len(metas) > 0:
+            s += [indent]
+            metas = [indent + tmp for tmp in metas]
+            s += metas
+
+        if len(subcols) > 0:
             s += [indent]
             subcols = [indent + tmp for tmp in subcols]
             s += subcols
 
-        return s
+        return '\n'.join(s)
+
+
+class _MetaSet(dict):
+    """
+    Manage a set of Meta
+    """
+    def __init__(self, coldir, verbose):
+        self._coldir = coldir
+        self._verbose = verbose
+
+    def __getitem__(self, name):
+        if name not in self:
+            raise RuntimeError(f'dict {name} not found')
+
+        return super().__getitem__(name)
+
+    def __setitem__(self, name, data):
+        raise TypeError(
+            f'To write to dict {name}, use meta[{name}].write(data) '
+            f'or meta[{name}].update(data)'
+        )
+
+    def _load(self, path):
+        name = util.extract_name(path)
+        if name in self:
+            raise RuntimeError(f'dict {name} already exists')
+
+        c = Meta(path, verbose=self._verbose)
+        super().__setitem__(name, c)
 
     def __repr__(self):
-        """
-        The columns representation to the world
-        """
-        s = self._get_repr_list()
-        s = '\n'.join(s)
-        return s
+        s = ['Metadata:']
+        s += ['  '+n for n in self]
+        return '\n'.join(s)
 
 
-def where(query_index):
-    """
-    Extract results from a query_index object into a normal numpy array.  This
-    is not usually necessary, as query objects inherit from numpy arrays.
+class _ColumnSubset(object):
+    def __init__(self, cols, columns):
+        self._cols = cols
+        self._columns = columns
 
-    Parameters
-    ----------
-    query_index: Indices
-        An Indices object generated by using operators such as "==" on an
-        indexed column object.  The Column methods between and match also
-        return Indices objects.  Indices objects can be combined with the "|"
-        and "&" operators.  This where functions extracts the underlying
-        index array.
-
-    Returns
-    -------
-    numpy array of indices
-
-    Example
-    --------
-
-        # lets say we have indexes on columns 'type' and 'mag' get the indices
-        # where type is 'event' and rate is between 10 and 20.  Note the braces
-        # around each operation are required here
-
-        >>> import columns
-        >>> c=columns.Columns(column_dir)
-        >>> ind=columns.where(  (c['type'] == 'event')
-                              & (c['rate'].between(10,20)) )
-
-        # now read some data from a set of columns using these
-        # indices.  We can do this with individual columns:
-        >>> ind = columns['id'][ind]
-        >>> x = columns['x'][ind]
-        >>> y = columns['y'][ind]
-
-        # we can also extract multiple columns at once
-        >>> data = columns.read(columns=['x','y','mag','type'], rows=ind)
-    """
-    return query_index.array()
+    def __getitem__(self, rows):
+        return self._cols.read(rows=rows, columns=self._columns)
diff --git a/pycolumns/convenience.py b/pycolumns/convenience.py
new file mode 100644
index 0000000..2bd8d76
--- /dev/null
+++ b/pycolumns/convenience.py
@@ -0,0 +1,127 @@
+import numpy as np
+from .columns import Columns
+from .defaults import DEFAULT_CACHE_MEM, DEFAULT_CHUNKSIZE
+from . import util
+
+
+def from_fits(
+    coldir,
+    filename,
+    ext=1,
+    native=False,
+    little=True,
+    lower=False,
+    compression=None,
+    chunksize=DEFAULT_CHUNKSIZE,
+    cache_mem=DEFAULT_CACHE_MEM,
+    verbose=False,
+    yes=False,
+):
+    """
+    Create or append to a columns database, reading from the input fits file.
+
+    parameters
+    ----------
+    coldir: str
+        Columns directory
+    filename: string
+        Name of the file to read
+    ext: extension, optional
+        The FITS extension to read, numerical or string. default 1
+    native: bool, optional
+        FITS files are in big endian byte order.
+        If native is True, ensure the outpt is in native byte order.
+        Default False.
+    little: bool, optional
+        FITS files are in big endian byte order.
+        If little is True, convert to little endian byte order. Default
+        True.
+    lower: bool, optional
+        if set to True, lower-case all names.  Default False.
+    compression: list or dict, optional
+        Either
+            1. A list of names that get default compression
+               see defaults.DEFAULT_COMPRESSION
+            2. A dict with keys set to columns names, possibly with
+               detailed compression settings.
+    chunksize: dict, str or number
+        The chunksize info for compressed columns.
+        See TableSchema.from_array for a full explanation
+    cache_mem: str or number
+        Cache memory for index creation, default '1g' or one gigabyte.
+    verbose: bool, optional
+        If set to True, display information
+    yes: bool, optional
+        If set to True, do not prompt for confirmation when overwriting
+        an existing directory
+    """
+    import fitsio
+
+    if (native and np.little_endian) or little:
+        byteswap = True
+        if verbose:
+            print('byteswapping')
+    else:
+        byteswap = False
+
+    with fitsio.FITS(filename, lower=lower) as fits:
+        hdu = fits[ext]
+
+        one = hdu[0:0+1]
+
+        if byteswap:
+            util.byteswap_inplace(one)
+
+        cols = Columns.create(
+            coldir,
+            yes=yes,
+            cache_mem=cache_mem,
+            verbose=verbose,
+        )
+        cols.from_array(
+            one,
+            compression=compression,
+            chunksize=chunksize,
+            append=False,
+        )
+
+        nrows = hdu.get_nrows()
+        rowsize = one.itemsize
+
+        # step size in bytes
+        step_bytes = int(cols.cache_mem_gb * 1024**3)
+        # step size in rows
+        step = step_bytes // rowsize
+
+        nstep = nrows // step
+        nleft = nrows % step
+
+        if nleft > 0:
+            nstep += 1
+
+        if verbose:
+            print('Loading %s rows from file: %s' % (nrows, filename))
+
+        for i in range(nstep):
+
+            start = i * step
+            stop = (i + 1) * step
+
+            if stop > nrows:
+                # not needed, but use for printouts
+                stop = nrows
+
+            if verbose:
+                print(f'    {start}:{stop} of {nrows}')
+
+            data = hdu[start:stop]
+
+            if byteswap:
+                util.byteswap_inplace(data)
+
+            cols.append(data, verify=False)
+            del data
+
+    cols.verify()
+
+    return cols
diff --git a/pycolumns/defaults.py b/pycolumns/defaults.py
new file mode 100644
index 0000000..e228365
--- /dev/null
+++ b/pycolumns/defaults.py
@@ -0,0 +1,33 @@
+# flake8: noqa
+import numpy as np
+
+ALLOWED_EXTENSIONS = [
+    'array', 'meta', 'index', 'index1', 'sorted', 'cols',
+    'chunks', 'json',
+]
+
+
+DEFAULT_CACHE_MEM = '1g'
+
+DEFAULT_CNAME = 'zstd'
+DEFAULT_CLEVEL = 5
+DEFAULT_SHUFFLE = 'bitshuffle'
+
+DEFAULT_COMPRESSION = {
+    'cname': DEFAULT_CNAME,
+    'clevel': DEFAULT_CLEVEL,
+    'shuffle': DEFAULT_SHUFFLE,
+}
+
+# 1 megabyte
+DEFAULT_CHUNKSIZE = '1m'
+
+CHUNKS_DTYPE = np.dtype(
+    [
+        ('offset', 'i8'),
+        ('nbytes', 'i8'),
+        ('rowstart', 'i8'),
+        ('nrows', 'i8'),
+        ('is_external', bool),
+    ]
+)
diff --git a/pycolumns/indices.py b/pycolumns/indices.py
index 15fe7ca..4b1e371 100644
--- a/pycolumns/indices.py
+++ b/pycolumns/indices.py
@@ -11,6 +11,19 @@ class Indices(np.ndarray):
     and "|" operators.  These return the intersection or union of values in two
     Indices objects.
 
+    Parameters
+    ----------
+    init_data: array or Indices
+        The initial data
+    copy: bool, optional
+        If set to True, ensure a copy of the input data is made
+    is_sorted: bool, optional
+        If set to True, the input data are sorted
+    is_checked: bool, optional
+        If set to True, the input rows have been checked to be in bounds,
+        and negatives are converted to positives.  This can be set later
+        with the .is_checked setter
+
     Methods:
         The "&" and "|" operators are defined.
 
@@ -25,18 +38,52 @@ class Indices(np.ndarray):
         Indices([3, 4, 5, 6])
 
     """
-    def __new__(self, init_data, copy=False):
-        self._is_sorted = False
+    def __new__(
+        self, init_data, copy=False, is_sorted=False, is_checked=False
+    ):
 
-        # always force native byte order since we send this to C code
-        # when using fitsio
+        # always force i8 and native byte order since we send this to C code
         arr = np.array(init_data, dtype='i8', copy=copy)
         shape = arr.shape
 
         ret = np.ndarray.__new__(self, shape, arr.dtype,
                                  buffer=arr)
+
+        self._is_sorted = is_sorted
+        if arr.ndim == 0:
+            self._is_sorted = True
+
+        self._is_checked = is_checked
+
         return ret
 
+    def get_minmax(self):
+        if self.ndim == 0:
+            mm = int(self), int(self)
+        else:
+
+            if self.is_sorted:
+                imin, imax = 0, self.size - 1
+            else:
+                s = self.sort_index
+                imin, imax = s[0], s[-1]
+
+            mm = self[imin], self[imax]
+
+        return mm
+
+    @property
+    def sort_index(self):
+        """
+        get an array that sorts the index
+        """
+        if self.is_sorted:
+            return None
+        else:
+            if not hasattr(self, '_sort_index'):
+                self._sort_index = self.argsort()
+            return self._sort_index
+
     @property
     def is_sorted(self):
         """
@@ -44,6 +91,22 @@ def is_sorted(self):
         """
         return self._is_sorted
 
+    @property
+    def is_checked(self):
+        """
+        returns True if the rows have been checked for negatives and
+        fixed
+        """
+        return self._is_checked
+
+    @is_checked.setter
+    def is_checked(self, val):
+        """
+        returns True if the rows have been checked for negatives and
+        fixed
+        """
+        self._is_checked = val
+
     def sort(self):
         """
         sort and set the is_sorted flag
@@ -51,6 +114,7 @@ def sort(self):
         if not self.is_sorted:
             if self.ndim > 0:
                 super(Indices, self).sort()
+            self._sort_index = None
             self._is_sorted = True
 
     def array(self):
@@ -62,7 +126,8 @@ def __and__(self, ind):
             w = np.intersect1d(self, ind)
         else:
             raise ValueError("comparison index must be an Indices object")
-        return Indices(w)
+
+        return Indices(w, is_sorted=True)
 
     def __or__(self, ind):
         # take the unique union
@@ -71,9 +136,16 @@ def __or__(self, ind):
         else:
             raise ValueError("comparison index must be an Indices object")
 
-        return Indices(w)
+        return Indices(w, is_sorted=True)
 
     def __repr__(self):
-        rep = np.ndarray.__repr__(self)
-        rep = rep.replace('array', 'Indices')
-        return rep
+        arep = np.ndarray.__repr__(self)
+        arep = arep.replace('array', 'Indices')
+
+        rep = [
+            'Indices:',
+            f'    size: {self.size}',
+            f'    sorted: {self.is_sorted}',
+            arep,
+        ]
+        return '\n'.join(rep)
diff --git a/pycolumns/mergesort.py b/pycolumns/mergesort.py
index b8dcdbf..32b78ec 100644
--- a/pycolumns/mergesort.py
+++ b/pycolumns/mergesort.py
@@ -1,16 +1,20 @@
 
-def create_mergesort_index(source, outfile, chunksize, tmpdir, verbose=False):
+def create_mergesort_index(
+    source,
+    isink,
+    ssink,
+    chunksize,
+    tmpdir,
+    verbose=False,
+):
     import tempfile
     import numpy as np
-    import fitsio
-    from .column import get_index_dtype
+    from ._column import Column as CColumn
     import time
 
     atm0 = time.time()
 
     nrows = source.nrows
-    dtype = source.dtype
-    index_dtype = get_index_dtype(dtype)
 
     nchunks = nrows // chunksize
     nleft = nrows % chunksize
@@ -21,7 +25,8 @@ def create_mergesort_index(source, outfile, chunksize, tmpdir, verbose=False):
     cache_size = chunksize // nchunks
 
     if verbose:
-        print('index_dtype:', index_dtype)
+        print('dtype:', source.dtype)
+        print('index_dtype:', source.index_dtype)
         print('nrows:', nrows)
         print('chunksize:', chunksize)
         print('cache size:', cache_size)
@@ -50,7 +55,7 @@ def create_mergesort_index(source, outfile, chunksize, tmpdir, verbose=False):
         # might be fewer than requested in slice
         chunk_num = chunk_value_data.size
 
-        chunk_data = np.zeros(chunk_num, dtype=index_dtype)
+        schunk_data = np.zeros(chunk_num, dtype=source.index_dtype)
 
         if verbose:
             print('argsort ', end='', flush=True)
@@ -60,10 +65,8 @@ def create_mergesort_index(source, outfile, chunksize, tmpdir, verbose=False):
         if verbose:
             _ptime(tm0sort)
 
-        # chunk_data['index'] = np.arange(start, start+chunk_num)
-        # chunk_data['value'] = chunk_value_data
-        chunk_data['index'] = np.arange(start, start+chunk_num)[s]
-        chunk_data['value'] = chunk_value_data[s]
+        ichunk_data = np.arange(start, start+chunk_num)[s]
+        schunk_data = chunk_value_data[s]
         del s
         del chunk_value_data
 
@@ -71,95 +74,108 @@ def create_mergesort_index(source, outfile, chunksize, tmpdir, verbose=False):
             print('writing ', end='', flush=True)
 
         tm0write = time.time()
-        tmpf = tempfile.mktemp(dir=tmpdir, suffix='.sf')
-        fobj = fitsio.FITS(tmpf, 'rw')
-        fobj.write(chunk_data)
+        itmpf = tempfile.mktemp(dir=tmpdir, suffix='.index')
+        ifobj = CColumn(itmpf, dtype=source.index_dtype, mode='w+')
+        ifobj.append(ichunk_data)
+
+        stmpf = tempfile.mktemp(dir=tmpdir, suffix='.sort')
+        sfobj = CColumn(stmpf, dtype=source.dtype, mode='w+')
+        sfobj.append(schunk_data)
 
         if verbose:
             _ptime(tm0write)
 
         # cache from lower chunk of data
-        cache = chunk_data[:cache_size].copy()
-        del chunk_data
+        icache = ichunk_data[:cache_size].copy()
+        scache = schunk_data[:cache_size].copy()
+
+        del ichunk_data
+        del schunk_data
 
         data = {
             # location where we will start next cache chunk
-            'main_index': cache.size,
+            'main_index': icache.size,
             'cache_index': 0,
-            'fits': fobj,
-            'cache': cache,
+            'ifile': ifobj,
+            'sfile': sfobj,
+            'icache': icache,
+            'scache': scache,
         }
         mergers.append(data)
         if verbose:
             _ptime(tm0, end='\n')
 
     # merge onto sink
-    stack_tops = np.zeros(len(mergers), dtype=index_dtype)
-    scratch = np.zeros(chunksize, dtype=index_dtype)
+
+    stack_itops = np.zeros(len(mergers), dtype=source.index_dtype)
+    stack_stops = np.zeros(len(mergers), dtype=source.dtype)
+    iscratch = np.zeros(chunksize, dtype=source.index_dtype)
+    sscratch = np.zeros(chunksize, dtype=source.dtype)
 
     for i, data in enumerate(mergers):
-        stack_tops[i] = data['cache'][data['cache_index']]
+        stack_itops[i] = data['icache'][data['cache_index']]
+        stack_stops[i] = data['scache'][data['cache_index']]
         data['cache_index'] += 1
 
-    iscratch = 0
+    scratch_ind = 0
 
     if verbose:
         print('Doing mergesort')
 
     mtm0 = time.time()
-    with fitsio.FITS(outfile, mode='rw', clobber=True) as sink:
-        nwritten = 0
-        it = 0
-        while len(mergers) > 0:
-            it += 1
-            imin = stack_tops['value'].argmin()
-
-            scratch[iscratch] = stack_tops[imin]
-            iscratch += 1
-
-            data = mergers[imin]
-
-            copy_new_top = True
-            if data['cache_index'] == data['cache'].size:
-
-                if data['main_index'] == data['fits'][1].get_nrows():
-                    # there is no more data left for this chunk
-                    ind = [i for i in range(stack_tops.size) if i != imin]
-                    stack_tops = stack_tops[ind]
-
-                    mergers[imin]['fits'].close()
-                    del mergers[imin]
-                    copy_new_top = False
-                else:
-                    # we have more data, lets load some into the cache
-                    main_index = data['main_index']
-                    next_index = main_index + cache_size
-                    data['cache'] = (
-                        data['fits'][1][main_index:next_index]
-                    )
-                    data['cache_index'] = 0
-                    data['main_index'] = main_index + data['cache'].size
-
-            if copy_new_top:
-                stack_tops[imin] = data['cache'][data['cache_index']]
-                data['cache_index'] += 1
-
-            if iscratch == scratch.size or len(mergers) == 0:
-                num2write = iscratch
-                nwritten += num2write
-                if verbose:
-                    print(f'\nwriting {nwritten}/{nrows}')
-
-                if len(sink) == 1:
-                    # nothing written yet
-                    sink.write(scratch[:num2write])
-                else:
-                    sink[1].append(scratch[:num2write])
-
-                iscratch = 0
+
+    nwritten = 0
+    it = 0
+    while len(mergers) > 0:
+        it += 1
+        imin = stack_stops.argmin()
+
+        iscratch[scratch_ind] = stack_itops[imin]
+        sscratch[scratch_ind] = stack_stops[imin]
+        scratch_ind += 1
+
+        data = mergers[imin]
+
+        copy_new_top = True
+        if data['cache_index'] == data['icache'].size:
+
+            if data['main_index'] == data['ifile'].nrows:
+                # there is no more data left for this chunk
+                ind = [i for i in range(stack_itops.size) if i != imin]
+                stack_itops = stack_itops[ind]
+                stack_stops = stack_stops[ind]
+
+                mergers[imin]['ifile'].close()
+                mergers[imin]['sfile'].close()
+                del mergers[imin]
+                copy_new_top = False
             else:
-                if verbose and it % (chunksize // 20) == 0:
-                    print('.', end='', flush=True)
+                # we have more data, lets load some into the cache
+                main_index = data['main_index']
+                next_index = main_index + cache_size
+                data['icache'] = data['ifile'][main_index:next_index]
+                data['scache'] = data['sfile'][main_index:next_index]
+                data['cache_index'] = 0
+                data['main_index'] = main_index + data['icache'].size
+
+        if copy_new_top:
+            stack_itops[imin] = data['icache'][data['cache_index']]
+            stack_stops[imin] = data['scache'][data['cache_index']]
+            data['cache_index'] += 1
+
+        if scratch_ind == iscratch.size or len(mergers) == 0:
+            num2write = scratch_ind
+            nwritten += num2write
+            if verbose:
+                print(f'\nwriting {nwritten}/{nrows}')
+
+            isink.append(iscratch[:num2write])
+            ssink.append(sscratch[:num2write])
+
+            scratch_ind = 0
+        else:
+            if verbose and it % (chunksize // 20) == 0:
+                print('.', end='', flush=True)
 
     if verbose:
         print('merge time: %.3g min' % ((time.time() - mtm0)/60))
diff --git a/pycolumns/metafile.py b/pycolumns/metafile.py
new file mode 100644
index 0000000..784b53a
--- /dev/null
+++ b/pycolumns/metafile.py
@@ -0,0 +1,99 @@
+import os
+from . import util
+
+
+class Meta(object):
+    """
+    Represent metadata that can be stored in JSON
+    """
+    def __init__(self, filename, verbose=False):
+        self._type = 'meta'
+        self._verbose = verbose
+        self._filename = filename
+        self._name = util.extract_name(filename)
+        self._dir = os.path.dirname(filename)
+
+    @property
+    def verbose(self):
+        return self._verbose
+
+    @property
+    def name(self):
+        """
+        get the name of this object
+        """
+        return self._name
+
+    @property
+    def dir(self):
+        """
+        get the directory holding the file
+        """
+        return self._dir
+
+    @property
+    def type(self):
+        """
+        get the type of this object
+        """
+        return self._type
+
+    @property
+    def filename(self):
+        """
+        get the file name holding the data
+        """
+        return self._filename
+
+    def write(self, data):
+        """
+        Write data
+
+        Parameters
+        ----------
+        data: Any json supported object
+            The data must be supported by the JSON format.
+        """
+
+        util.write_json(self.filename, data)
+
+    def update(self, data):
+        """
+        Update the data.  This only works for dictionaries
+
+        Parameters
+        ----------
+        data: dict
+            Update the data.  The stored data and the input must be dict or
+            dict like
+        """
+
+        odata = self.read()
+        odata.update(data)
+        util.write_json(self.filename, odata)
+
+    def read(self):
+        """
+        read the data
+        """
+        return util.read_json(self.filename)
+
+    def __repr__(self):
+        """
+        Get a list of metadat for this column.
+        """
+        indent = '  '
+
+        s = []
+        if self.name is not None:
+            s += ['name: %s' % self.name]
+
+        if self.filename is not None:
+            s += ['filename: %s' % self.filename]
+
+        s += ['type: meta']
+
+        s = [indent + tmp for tmp in s]
+        s = ['Meta: '] + s
+
+        return '\n'.join(s)
diff --git a/pycolumns/schema.py b/pycolumns/schema.py
new file mode 100644
index 0000000..9bf523b
--- /dev/null
+++ b/pycolumns/schema.py
@@ -0,0 +1,335 @@
+import numpy as np
+from . import util
+from .defaults import DEFAULT_COMPRESSION, DEFAULT_CHUNKSIZE
+
+
+class TableSchema(dict):
+    """
+    A table Schema
+
+    Parameters
+    ----------
+    schemas: [ColumnSchema], optional
+        A list of ColumnSchema to be added to the table schema.
+        You can add more using schema.add_column(schema)
+
+    Examples
+    --------
+
+    # Construct from ColumnSchema objects
+    cslist = [
+        ColumnSchema(name='id', dtype='i8', compression=True),
+        ColumnSchema(name='ra', dtype='f8'),
+        ColumnSchema(name='name', dtype='U5',
+                     compression={'shuffle': 'shuffle'},
+                     chunksize='0.5m'),
+    ]
+    schema = TableSchema(cslist)
+
+    # Or you can add one by one
+    schema = TableSchema(cslist)
+    for cs in cslist:
+        schema.add(cs)
+
+    # One can also construct TableSchema from dicts, other TableSchema and
+    # arrays (with fields) using class methods
+
+    schema = TableSchema.from_array(array, compression=c, chunksize=ch)
+    schema = TableSchema.from_schema(some_dict)
+    schema = TableSchema.from_columns(cslist)  # same as using constructor
+
+    Layout
+    --------
+    # A TableSchema inherits from a dict. It may look something like this
+    {
+        'id': {'dtype': '<i8'}
+        'ra': {'dtype': '<f8'}
+        'dec': {'dtype': '<f8'}
+        'name': {'dtype': '<U5'}
+    }
+
+    # Compression and chunksize for compressed files can be set
+    {
+        'id': {
+            'dtype': '<i8',
+            'compression': {
+                'cname': 'zlib',
+                'clevel': 5,
+                'shuffle': 'bitshuffle',
+            }
+            'chunksize': '1m',
+        },
+        'ra': {'dtype': '<f8'}
+    }
+    """
+    def __init__(self, schemas=None):
+        if schemas is None:
+            return
+
+        for schema in schemas:
+            self.add_column(schema)
+
+    def add_column(self, schema):
+        """
+        add a column schema
+        """
+        self[schema.name] = schema.copy()
+
+    @classmethod
+    def from_array(
+        cls, array, compression=None, chunksize=DEFAULT_CHUNKSIZE,
+        fill_value=None,
+    ):
+        """
+        Convert an array with fields to a TableSchema
+
+        Parameters
+        ----------
+        array: numpy array with fields or dict of arrays
+            An array with fields defined, or a dict holding arrays
+        compression: dict, list, bool, or None, optional
+            - If None or False, do not set compression
+            - If True, use default compression
+            - If a list, return use default compression if the name is in the
+              list
+            - If a dict, and name is an entry
+                 - if value is True, use default compression
+                 - if value is a dict, return the dict with defaults set for
+                   non specified parameters
+        chunksize: dict, str or number
+            A dict or str or number
+
+            - if str or number, it is applied to all compressed columns.
+            - a dict,  keyed by name, it gives chunksize for certain columns.
+        fill_value: dict, optional
+            Optional dict specifying fill values for columns.
+
+        Returns
+        -------
+        TableSchema
+        """
+        names = util.get_data_names(array)
+
+        table_schema = TableSchema()
+        for name in names:
+            keys = {}
+            comp = _get_column_compression(compression, name)
+            if comp:
+                # Note, if comp is a dict, it gets defaults set for
+                # non specified compression entries in ColumnSchema
+                keys['compression'] = comp
+                keys['chunksize'] = _get_column_chunksize(chunksize, name)
+
+            if fill_value is not None and name in fill_value:
+                keys['fill_value'] = fill_value[name]
+
+            schema = ColumnSchema(
+                name=name,
+                dtype=array[name].dtype,
+                **keys
+            )
+            table_schema.add_column(schema)
+
+        return table_schema
+
+    @classmethod
+    def from_schema(cls, sdict):
+        """
+        Convert a TableSchema or dict to a TableSchema
+
+        Parameters
+        ----------
+        sdict: dict or TableSchema
+            e.g.
+            {
+              'id': {
+                'dtype': 'f8',
+                'compression': True,
+              },
+              'ra': {'dtype': 'f8'},
+              'nn': {
+                  'dtype': 'U3',
+                  'fill_value': '-',
+              },
+            }
+
+        Returns
+        -------
+        TableSchema
+        """
+
+        table_schema = TableSchema()
+        for name, dschema in sdict.items():
+
+            schema = ColumnSchema(name=name, **dschema)
+            table_schema.add_column(schema)
+
+        return table_schema
+
+    @classmethod
+    def from_columns(cls, schemas):
+        """
+        Construct a new table Schema from a list of ColumnSchema objects
+
+        Parameters
+        ----------
+        schemas: [ColumnSchema], optional
+            A list of ColumnSchema to be added to the table schema.
+            You can add more using schema.add_column(schema)
+        """
+        return TableSchema(schemas)
+
+
+class ColumnSchema(dict):
+    """
+    A schema for a column
+
+    Parameters
+    ----------
+    dtype: numpy dtype
+        A numpy dtype for the column
+    compression: dict, bool, or None, optional
+        If None (the default) or False, no compression is used.
+        If True, the default compression settings are used.
+
+        Otherwise a dict describing the compression.  Any missing
+        entries are filled in with defaults from pycolumns.DEFAULT_COMPRESSION
+        compression = {
+            'cname': 'zstd',
+            'clevel': 5,
+            'shuffle': 'bitshuffle',
+        }
+    chunksize: str or number, optional
+        The chunksize for compressed columns.  Default is
+        pycolumns.DEFAULT_CHUNKSIZE.
+
+        This describes the size of the uncompressed chunk.  Can be a string
+        with units or a number, in which case it is interpreted as bytes
+
+            '1m': 1 megabyte
+            '2k': 2 kilobytes
+            '0.5g': 0.5 gigabytes
+            '200b': 200 bytes
+            '1000r': 1000 rows
+    fill_value: value, optional
+        If sent and not None, resized columns get this fill value.  Default
+        is zero/empty string (bit zeros)
+    """
+    def __init__(
+        self,
+        name,
+        dtype,
+        compression=None,
+        chunksize=DEFAULT_CHUNKSIZE,
+        fill_value=None,
+    ):
+        self._name = name
+        schema = _make_array_schema_dict(
+            dtype=dtype,
+            compression=compression,
+            chunksize=chunksize,
+            fill_value=fill_value,
+        )
+        self.update(schema)
+
+    @property
+    def name(self):
+        """
+        get the column name
+        """
+        return self._name
+
+
+def _make_array_schema_dict(
+    dtype,
+    compression=None,
+    chunksize=DEFAULT_CHUNKSIZE,
+    fill_value=None,
+):
+    schema = {
+        'dtype': np.dtype(dtype).str,
+    }
+    if fill_value is not None:
+        schema['fill_value'] = fill_value
+
+    if compression:
+        schema['chunksize'] = chunksize
+        schema['compression'] = DEFAULT_COMPRESSION.copy()
+        if compression is not True:
+            schema['compression'].update(compression)
+
+    return schema
+
+
+def _get_column_compression(compression, name):
+    """
+    extract compression info for a specified column
+
+    Parameters
+    ----------
+    compression: bool, dict or list
+        If a bool, return None for False or defaults for True
+        If a list, return defaults if the name is in the list
+        If a dict, return the entry or None
+    name: str
+        Column name
+
+    Returns
+    -------
+    None or a dict
+    """
+    if hasattr(compression, 'keys'):
+        tcomp = compression.get(name)
+        if tcomp:
+            comp = tcomp
+        else:
+            comp = None
+    elif _has_len(compression):
+        if name in compression:
+            comp = True
+        else:
+            comp = None
+    else:
+        if not compression:
+            comp = None
+        else:
+            comp = True
+
+    return comp
+
+
+def _get_column_chunksize(chunksize, name):
+    """
+    extract chunksize for a specified column
+
+    Parameters
+    ----------
+    chunksize: dict or str or number
+        If an str or number, just return
+        If a dict, return the entry if found or None
+    name: str
+        Column name
+
+    Returns
+    -------
+    str or number
+    """
+    if hasattr(chunksize, 'keys'):
+        return chunksize.get(name, DEFAULT_CHUNKSIZE)
+    else:
+        # catch a bug
+        if hasattr(chunksize, 'append') or hasattr(chunksize, 'discard'):
+            raise ValueError(
+                'expected dict, str, or number, got {type(chunksize)}'
+            )
+        return chunksize
+
+
+def _has_len(x):
+    try:
+        len(x)
+        ret = True
+    except TypeError:
+        ret = False
+
+    return ret
diff --git a/pycolumns/sfile.py b/pycolumns/sfile.py
deleted file mode 100644
index f0ad888..0000000
--- a/pycolumns/sfile.py
+++ /dev/null
@@ -1,722 +0,0 @@
-"""
-Read and write numpy arrays to a simple file format.  The format is a
-simple ascii header followed by data in binary format.
-
-TODO
-
-    - for full col read, or slices, use fromfile rather than mmap
-"""
-import os
-import pprint
-
-import numpy as np
-
-SIMPLEFILE_VERSION = '0.1'
-
-
-class SimpleFile(object):
-    """
-    This class implements a simple file format for holding numerical python
-    arrays.  The format is a simple ascii header followed by data in
-    binary form
-
-    The format is designed to represent a column of data, which can itself be
-    multi-dimensional.  This is not designed to represent a full rec array, for
-    that see e.g. fitsio or esutil.sfile
-
-    Examples
-    ---------
-
-    # Open and read from a .sf file
-    # the
-    with SimpleFile('test.sf') as sf:
-
-        # see what's in the file.
-        print(sf)
-        filename: 'test.sf'
-        mode: 'r'
-        size: 64526192
-        dtype: '<f4'
-        hdr:
-          {'dataset': 'dc4',
-           'pyvers':'v2.6.2'}
-
-        #
-        # A few ways to read all the data
-        #
-
-        data = sf[:]
-
-        # read a subset of rows.  Can use slices, single numbers for rows,
-        # or list/array of row numbers.
-
-        data = sf[35]
-        data = sf[35:100]
-        data = sf[ [35,66,22,23432] ]
-        data = sf[ row_list ]
-
-    with SimpleFile('test.sf','w+') as sf:
-        sf.write(data)
-
-        # this appends data
-        sf.write(more_data)
-
-        # check the first row of the data we have written
-        sf[0]
-
-        sf.write(more_data)
-    """
-    def __init__(self, filename=None, mode='r'):
-        self.open(filename=filename, mode=mode)
-
-    def open(self, filename=None, mode='r'):
-
-        self.close()
-
-        self._mode = mode
-        self._filename = filename
-
-        if filename is None:
-            return
-
-        # expand shortcut variables
-        fpath = os.path.expanduser(filename)
-        fpath = os.path.expandvars(fpath)
-        self._filename = fpath
-
-        if mode == 'r+' and not os.path.exists(self._filename):
-            # path doesn't exist but we want to append.  Change the
-            # mode to write
-            mode = 'w+'
-
-        self._fobj = open(self._filename, _fix_mode(self.mode))
-
-        if self._mode[0] == 'w':
-            self._is_empty = True
-        else:
-            self._is_empty = False
-            self._load_metadata()
-
-    def _load_metadata(self):
-        self._hdr = self._read_header()
-
-        self._shape = self._hdr['shape']
-        self._descr = self._hdr['dtype']
-        self._dtype = np.dtype(self._descr)
-
-        self._mmap = np.memmap(
-            self._filename,
-            dtype=self._dtype,
-            mode=self._mode,
-            offset=self._data_start,
-            shape=self._shape,
-        )
-
-    def close(self):
-        """
-        Close the file and reset the metadata to None
-        """
-
-        if hasattr(self, '_mmap'):
-            # official way to close the file according to docs
-            del self._mmap
-
-        if hasattr(self, '_fobj') and self._fobj is not None:
-            self._fobj.close()
-
-        self._filename = None
-        self._mode = None
-        self._fobj = None
-        self._mmap = None
-        self._hdr = None
-        self._data_start = None
-        self._shape = None
-        self._descr = None
-        self._dtype = None
-
-    @property
-    def mmap(self):
-        """
-        get a reference to the mmap
-        """
-        return self._mmap
-
-    @property
-    def size(self):
-        """
-        get the total number of elements
-        """
-        if self._mmap is None:
-            raise RuntimeError("no file has been opened for reading")
-
-        return self._mmap.size
-
-    @property
-    def itemsize(self):
-        """
-        get the total number of elements
-        """
-        return self._dtype.itemsize
-
-    @property
-    def shape(self):
-        """
-        get the shape of the data
-        """
-        return self._shape
-
-    @property
-    def dtype(self):
-        """
-        get the number of rows in the file
-        """
-        return self._dtype
-
-    @property
-    def header(self):
-        """
-        get a copy of the header
-        """
-        if self._hdr is None:
-            raise RuntimeError("no file has been opened for reading")
-        hdr = {}
-        hdr.update(self._hdr)
-        return hdr
-
-    @property
-    def mode(self):
-        """
-        get the file open mode
-        """
-        return self._mode
-
-    @property
-    def filename(self):
-        """
-        get the file name
-        """
-        return self._filename
-
-    def write(self, data):
-        """
-        write data to the file, appending if the file is not empty
-
-        paramters
-        ---------
-        data: array
-            A structured numerical python array.  If data already
-            exists in the file, this data must have compatible
-            data type.  For binary files this includes the byte
-            order.
-        header: dict, optional
-            Optional dictionary to write into the header.  This
-            can only be written the first time.
-        """
-
-        self._ensure_open_for_writing()
-
-        fobj = self._fobj
-
-        if not self._is_empty:
-            # check compatible, in case there is already data in the file
-            self._ensure_compatible_dtype(data)
-        else:
-            self._write_initial_header(data)
-
-        # zero bytes from the end of the file
-        fobj.seek(0, 2)
-        data.tofile(fobj)
-
-        if not self._is_empty:
-            self._update_shape(data.shape)
-
-        self.open(
-            filename=self._filename,
-            mode='r+',
-        )
-
-    def __getitem__(self, arg):
-        """
-        # read subsets from the file.
-        sf = SimpleFile(....)
-
-        # read subsets of rows
-        data = sf[35]
-        data = sf[35:88]
-        data = sf[35:88, 3, 5]
-        data = sf[[3, 234, 5551]]
-        """
-
-        data = self._mmap[arg]
-
-        if isinstance(data, np.memmap):
-            # slices will return a view on the memmap
-            # this will force a copy
-            return np.array(data)
-        else:
-            return data
-
-    def __setitem__(self, arg, values):
-        """
-        write to the file
-
-        sf[35] = 10
-        sf[35:88] = ...
-        sf[[3, 234, 5551]] = [8, 9, 10]
-        """
-
-        self._mmap[arg] = values
-
-    def _ensure_open(self):
-        """
-        check if a file is open, if not raise a RuntimeError
-        """
-        if not hasattr(self, '_fobj') or self._fobj is None:
-            raise RuntimeError("no file is open")
-
-    def _ensure_open_for_writing(self):
-        """
-        check if a file is open for writing, if not raise a RuntimeError
-        """
-        self._ensure_open()
-        if self._fobj.mode[0] != 'w' and '+' not in self._mode:
-            raise ValueError("You must open with 'w*' or 'r+' to write")
-
-    def _ensure_open_for_reading(self):
-        """
-        check if a file is open for reading, if not raise a RuntimeError
-        """
-        self._ensure_open()
-
-        if self._fobj.mode[0] != 'r' and '+' not in self._mode:
-            raise ValueError("You must open with 'w+' or 'r*' to read")
-
-    def _ensure_compatible_dtype(self, data):
-        """
-        if we are writing binary we demand exact match.
-
-        For text we just make sure everything matches if the
-        byte order is ignored
-        """
-
-        # if self._dtype is not None, there was data in the file
-        if self._dtype is not None:
-            if self._dtype != data.dtype:
-                m = (
-                    "attempt to write an incompatible "
-                    "data type: expected '%s' got '%s'"
-                )
-                m = m % (self._descr, data.dtype.descr)
-                raise ValueError(m)
-
-    def _write_initial_header(self, data):
-        """
-        write the intial header for a new file
-        """
-        fobj = self._fobj
-        fobj.seek(0)
-
-        shape_string = _get_shape_string(data.shape)
-        dtype_string = _get_dtype_string(data.dtype)
-        version_string = _get_version_string()
-        header_end = _get_header_end()
-
-        fobj.write(shape_string)
-        fobj.write(dtype_string)
-        fobj.write(version_string)
-        fobj.write(header_end)
-        self._data_start = fobj.tell()
-
-    def _update_shape(self, shape_add):
-        """
-        update the shape entry in the file
-        """
-
-        fobj = self._fobj
-        fobj.seek(0)
-
-        shape_current = self.shape
-        if shape_current is None:
-            raise RuntimeError('Attempting to update shape but '
-                               'not found in header')
-
-        ndim = len(shape_current)
-        assert ndim == len(shape_add)
-        if ndim > 1:
-            # multi-dimensional array
-            for i in range(1, ndim):
-                assert shape_add[i] == shape_current[i]
-
-        nrows_new = shape_current[0] + shape_add[0]
-        shape_new = (nrows_new,)
-        if ndim > 1:
-            shape_new = shape_new + shape_current[1:]
-
-        shape_string = _get_shape_string(shape_new)
-        fobj.write(shape_string)
-        self._hdr['shape'] = shape_new
-
-    def _read_header(self):
-        """
-        Read the header from a simple self-describing file format with an
-        ascii header.
-
-        The file format:
-          First line:
-              SHAPE = --------------
-        where the shape can be, e.g. (125,) for a 1-d array, or (25,8,15) for
-        a 3-d array.  The -- pad it out to 40 characters
-
-        This exact formatting is required so SHAPE can be updated *in place*
-        when appending rows to a file.  Note the file can always be read as
-        long as the first line reads 'SHAPE = some_number' but appending
-        requires the exact format.
-
-        Last two lines of the header region must be:
-                END
-                blank line
-        case does not matter.
-
-        In between the SHAPE and END lines is the header data.  This is a
-        string that must eval() to a dictionary.  It must contain the
-        following entry:
-
-              DTYPE = 'f8'
-
-        Where any valid scalar dtype is ok.  There should also be a VERSION
-        tag.
-
-              VERSION = '0.1'
-
-        The rest of the keywords can by any variable can be used as long as it
-        can be eval()d.
-
-        An example header:
-            SHAPE = (1032, 2)
-            VERSION = 0.1
-            DTYPE = '>f8'
-            END
-
-            -- data begins --
-        """
-
-        fobj = self._fobj
-
-        if fobj is None:
-            raise ValueError('there is no open file')
-
-        fobj.seek(0)
-
-        lines = []
-        for line in fobj:
-            line = str(line, 'ascii').strip()
-
-            if line == 'END':
-                break
-
-            lines.append(line)
-
-        assert line == 'END', 'expected END line'
-
-        # we read one more blank line
-        tmpline = str(fobj.readline(), 'ascii').strip()
-        assert tmpline == '', 'expected blank line after END'
-
-        self._data_start = fobj.tell()
-        hdr = {
-            'shape': _extract_shape_from_string(lines[0]),
-            'dtype': _extract_dtype_from_string(lines[1]),
-            'version': _extract_version_from_string(lines[2]),
-        }
-
-        return hdr
-
-    def __repr__(self):
-
-        top = ["filename: '%s'" % self._filename]
-
-        s = ["mode: '%s'" % self._mode]
-
-        s += ["shape: %s" % (self._shape,)]
-
-        if self._descr is not None:
-            drepr = pprint.pformat(self._descr).split('\n')
-            drepr = ['  '+d for d in drepr]
-
-            s += ["dtype:"]
-            s += drepr
-
-        slist = []
-        for tmp in s:
-            slist.append('    '+tmp)
-
-        slist = top + slist
-        rep = "\n".join(slist)
-        return rep
-
-    def __enter__(self):
-        return self
-
-    def __exit__(self, exception_type, exception_value, traceback):
-        self.close()
-
-
-def read(fname, rows=None):
-    """
-    Read a numpy array from a simple self-describing file format with an ascii
-    header.
-
-    Parameters
-    ----------
-    fname: string
-        The name of the file to read
-    rows: optional
-        Optional rows to read
-    """
-
-    with SimpleFile(fname) as sf:
-        if rows is not None:
-            data = sf[rows]
-        else:
-            data = sf[:]
-
-    return data
-
-
-def write(fname, data, append=False):
-    """
-    Write a numpy array into a simple self-describing file format with an ascii
-    header.
-
-    Parameters
-    ----------
-    fname: string
-        The filename to write
-    data: array
-        Numerical python array, a structured array with fields
-    append: bool, optional
-    """
-
-    if append:
-        mode = 'r+'
-    else:
-        mode = 'w'
-
-    with SimpleFile(fname, mode=mode) as sf:
-        sf.write(data)
-
-
-class Slicer(SimpleFile):
-    """
-    This version doesn't use memmap and thus only can do slices
-    Still supports writing/appending
-    """
-    def _load_metadata(self):
-        self._hdr = self._read_header()
-
-        self._shape = self._hdr['shape']
-        self._descr = self._hdr['dtype']
-        self._dtype = np.dtype(self._descr)
-        self._set_nrows()
-
-    def _set_nrows(self):
-        self._fobj.seek(0, 2)
-        data_size_bytes = self._fobj.tell() - self._data_start
-        self._nrows = data_size_bytes // self.itemsize
-        if data_size_bytes % self.itemsize != 0:
-            raise ValueError(
-                f'Data size {data_size_bytes} bytes not multiple '
-                f'of itemsize {self.itemsize}'
-            )
-
-    @property
-    def mmap(self):
-        raise ValueError('Slicer has no mmap')
-
-    @property
-    def nrows(self):
-        """
-        get the total number of rows
-        """
-
-        return self._nrows
-
-    @property
-    def size(self):
-        """
-        get the total number of elements
-        """
-
-        return self.nrows
-
-    @property
-    def shape(self):
-        """
-        get the shape of the data
-        """
-        return (self.nrows, )
-
-    @property
-    def dtype(self):
-        """
-        get the number of rows in the file
-        """
-        return self._dtype
-
-    @property
-    def header(self):
-        """
-        get a copy of the header
-        """
-        if self._hdr is None:
-            raise RuntimeError("no file has been opened for reading")
-        hdr = {}
-        hdr.update(self._hdr)
-        return hdr
-
-    def __getitem__(self, index):
-        import numpy as np
-
-        if isinstance(index, slice):
-            is_scalar = False
-            start = index.start
-            count = index.stop - index.start
-        else:
-            is_scalar = True
-            start = index
-            count = 1
-
-        self._fobj.seek(self._data_start + start * self.itemsize)
-        data = np.fromfile(self._fobj, dtype=self.dtype, count=count)
-        if is_scalar:
-            data = data[0]
-
-        return data
-
-
-def _get_shape_string(shape):
-    """
-    Specially formatted fixed-length for updating later
-    """
-    s = 'SHAPE = %40s\n' % (shape,)
-    return bytes(s, 'ascii')
-
-
-def _get_dtype_string(dtype):
-    """
-    dtype string for header
-    """
-    if dtype.names is not None:
-        dstr = str(dtype.descr)
-        s = "DTYPE = %s\n" % dstr
-    else:
-        dstr = dtype.descr[0][1]
-        s = "DTYPE = '%s'\n" % dstr
-    return bytes(s, 'ascii')
-
-
-def _get_version_string():
-    """
-    version string for header
-    """
-    s = "VERSION = '%s'\n" % SIMPLEFILE_VERSION
-    return bytes(s, 'ascii')
-
-
-def _get_header_end():
-    """
-    version string for header
-    """
-    s = "END\n\n"
-    return bytes(s, 'ascii')
-
-
-def _extract_shape_from_string(line):
-    lsplit = line.split('=')
-    if len(lsplit) != 2:
-        raise ValueError("First line of header must be SHAPE = %40d")
-    fname = lsplit[0].strip()
-
-    # also allow old NROWS word for compatibility
-    if fname.upper() != 'SHAPE':
-        raise ValueError("First line of header must be SHAPE = %40d")
-
-    shape = eval(lsplit[1])
-
-    return shape
-
-
-def _extract_dtype_from_string(line):
-    lsplit = line.split('=')
-    if len(lsplit) != 2:
-        raise ValueError("Second line of header must be DTYPE = ...")
-    fname = lsplit[0].strip()
-
-    # also allow old NROWS word for compatibility
-    if fname.upper() != 'DTYPE':
-        raise ValueError("Second line of header must be DTYPE = ...")
-
-    dtype = eval(lsplit[1])
-
-    return dtype
-
-
-def _extract_version_from_string(line):
-    lsplit = line.split('=')
-    if len(lsplit) != 2:
-        raise ValueError("Third line of header must be VERSION = ...")
-    fname = lsplit[0].strip()
-
-    # also allow old NROWS word for compatibility
-    if fname.upper() != 'VERSION':
-        raise ValueError("Third line of header must be VERSION = ...")
-
-    version = eval(lsplit[1])
-
-    return version
-
-
-_PYMODES = ['rb', 'r+b', 'wb', 'w+b']
-
-
-def _fix_mode(mode):
-    if mode in _PYMODES:
-        return mode
-
-    if mode == 'r':
-        mode = 'rb'
-    elif mode == 'r+':
-        mode = 'r+b'
-    elif mode == 'w':
-        mode = 'wb'
-    elif mode == 'w+':
-        mode = 'w+b'
-    else:
-        raise ValueError("bad open mode: '%s'" % mode)
-
-    return mode
-
-
-def test():
-    """
-    very simple test
-    """
-    import tempfile
-    with tempfile.TemporaryDirectory() as tmpdir:
-        fname = os.path.join(tmpdir, 'test.sf')
-        data = np.array([[1.0, 3.5],
-                         [4.5, 2.6],
-                         [8.7, 1.5]])
-
-        with SimpleFile(fname, mode='w+') as sf:
-            sf.write(data)
-            sf.write(data)
-
-            rdata = sf[:]
-            assert np.all(data == rdata[0:3])
-            assert np.all(data == rdata[3:])
-
-        with SimpleFile(fname) as sf:
-            rdata = sf[:]
-            assert np.all(data == rdata[0:3])
-            assert np.all(data == rdata[3:])
diff --git a/pycolumns/util.py b/pycolumns/util.py
index 8a51abb..99c5cb1 100644
--- a/pycolumns/util.py
+++ b/pycolumns/util.py
@@ -1,9 +1,9 @@
 import os
-import json
 import numpy as np
+from . import defaults
 
 
-def extract_rows(rows, sort=True):
+def extract_rows(rows, nrows, check_slice_stop=False):
     """
     extract rows for reading
 
@@ -11,87 +11,202 @@ def extract_rows(rows, sort=True):
     ----------
     rows: sequence, Indices, slice or None
         Possible rows to extract
-    sort: bool, optional
-        Whether to sort when converted to Indices
+    nrows: int
+        Total number of rows
+    check_slice_stop: bool, optional
+        this is for doing row updates and when we need
+        the slice to be exact, not go beyond nrows
 
     Returns
     -------
-    Indices, possibly sorted.  Note this does not make a copy of the data
+    Indices, or slice.  possibly sorted.
     """
     from .indices import Indices
 
-    if (
-        rows is not None
-        and not isinstance(rows, slice)
-        and not isinstance(rows, Indices)
-    ):
-        output = Indices(rows)
-    else:
+    if isinstance(rows, Indices) and rows.is_checked:
+        return rows
+
+    if isinstance(rows, slice):
+        s = extract_slice(rows, nrows, check_slice_stop=check_slice_stop)
+        if s.step is not None:
+            ind = np.arange(s.start, s.stop, s.step)
+            output = Indices(ind, is_sorted=True)
+        else:
+            output = s
+
+    elif rows is None:
+        output = slice(0, nrows)
+    elif isinstance(rows, Indices):
         output = rows
+    else:
+        output = Indices(rows)
 
-    if isinstance(rows, Indices) and sort:
-        output.sort()
+    if isinstance(output, Indices):
+        if output.ndim == 0 and output < 0:
+            output = Indices(nrows + output, is_checked=True)
+        else:
+            w, = np.where(output < 0)
+            if w.size > 0:
+                # make a copy since we don't want to modify underlying
+                # input data
+                output = output.copy()
+                output[w] += nrows
+                output.is_checked = True
+                assert output.is_checked
 
     return output
 
 
-def extract_colname(filename):
+def extract_slice(s, nrows, check_slice_stop=False):
+    start = s.start
+    stop = s.stop
+
+    if stop is not None and check_slice_stop:
+        # this is for doing row updates and when we need
+        # the slice to be exact, not go beyond nrows
+        if stop > nrows:
+            raise IndexError(f'slice stop {stop} > nrows {nrows}')
+
+    if start is None:
+        start = 0
+    if stop is None:
+        stop = nrows
+
+    if start < 0:
+        start = nrows + start
+        if start < 0:
+            raise IndexError("Index out of bounds")
+
+    if stop < 0:
+        stop = nrows + stop
+
+    if stop < start:
+        # will return an empty struct
+        stop = start
+
+    if stop is None or stop > nrows:
+        stop = nrows
+
+    return slice(start, stop, s.step)
+
+
+def extract_name(filename):
     """
     Extract the column name from the file name
     """
 
-    bname = os.path.basename(filename)
-    name = '.'.join(bname.split('.')[0:-1])
+    name, _ = split_ext(filename)
     return name
+    # n, ext = split_ext(filename)
+    # if ext == 'cols':
+    #     return f'/{n}'
+    # else:
+    #     return n
+    # bname = os.path.basename(filename)
+    # name = '.'.join(bname.split('.')[0:-1])
+    # return name
 
 
-def extract_coltype(filename):
+def extract_type(filename):
     """
     Extract the type from the file name
     """
-    return filename.split('.')[-1]
+    return extract_extension(filename)
 
 
-def create_filename(dir, name, type):
+def extract_extension(filename):
     """
-    genearte a file name from dir, column name and column type
+    Extract the extension
     """
-    if dir is None:
-        raise ValueError('Cannot create column filename, dir is None')
+    _, ext = split_ext(filename)
+    return ext
 
-    if name is None:
-        raise ValueError('Cannot create column filename: name is None')
 
-    if type is None:
-        raise ValueError('Cannot create column filename: type is None')
+def split_ext(filename):
+    """
+    For /path/to/blah.txt returns ('blah', 'txt')
+    """
+    bname = os.path.basename(filename)
+    s = bname.split('.')
+    if len(s) == 1:
+        name = s[0]
+        ext = ''
+    else:
+        name = '.'.join(s[0:-1])
+        ext = s[-1]
+    return name, ext
 
-    if type == 'dict':
-        ext = 'json'
-    elif type == 'array':
-        ext = 'array'
+
+def get_meta_filename(path):
+    """
+    Get a path to a .meta file assuming path is a column directory
+    """
+    bname = os.path.basename(path)
+    return f'{path}/{bname}.meta'
+
+
+def is_column(path):
+    """
+    Returns True if path is a directory (or link) containing
+    a file named basename(path).meta
+    """
+    metapath = get_meta_filename(path)
+    if os.path.exists(metapath):
+        return True
     else:
-        raise ValueError("bad file type: '%s'" % type)
+        return False
 
-    return os.path.join(dir, name+'.'+ext)
+
+def get_column_dir(dir, name):
+    """
+    get path to column directory
+    """
+    return os.path.join(dir, name)
+
+
+def get_filename(dir, name, ext):
+    """
+    genearte a file name from dir, column name and column type
+    """
+    if ext not in defaults.ALLOWED_EXTENSIONS:
+        raise ValueError(f'unsuported extension {ext}')
+
+    return os.path.join(dir, f'{name}.{ext}')
+
+
+def get_colfiles(coldir):
+    name = os.path.basename(coldir)
+    return {
+        'dir': coldir,
+        'name': name,
+        'meta': get_filename(coldir, name, 'meta'),
+        'array': get_filename(coldir, name, 'array'),
+        'index': get_filename(coldir, name, 'index'),
+        'index1': get_filename(coldir, name, 'index1'),
+        'sorted': get_filename(coldir, name, 'sorted'),
+        'chunks': get_filename(coldir, name, 'chunks'),
+    }
 
 
 def read_json(fname):
     """
     wrapper to read json
     """
+    import json
 
     with open(fname) as fobj:
         data = json.load(fobj)
     return data
 
 
-def write_json(obj, fname, pretty=True):
+def write_json(fname, obj):
     """
     wrapper for writing json
     """
+    import json
 
     with open(fname, 'w') as fobj:
-        json.dump(obj, fobj, indent=1, separators=(',', ':'))
+        json.dump(obj, fobj, indent=4, separators=(',', ':'))
 
 
 def get_native_data(data):
@@ -121,33 +236,222 @@ def get_native_data(data):
     return new_data
 
 
-def maybe_decode_fits_ascii_strings_to_unicode_py3(array):
-    new_dtype, do_conversion = (
-        maybe_convert_ascii_dtype_to_unicode(array.dtype)
-    )
-    if do_conversion:
-        array = array.astype(new_dtype, copy=False)
-    return array
-
-
-def maybe_convert_ascii_dtype_to_unicode(dtype):
-
-    do_conversion = False
-    new_dt = []
-    for dt in dtype.descr:
-        if 'S' in dt[1]:
-            do_conversion = True
-            if len(dt) == 3:
-                new_dt.append((
-                    dt[0],
-                    dt[1].replace('S', 'U').replace('|', ''),
-                    dt[2]))
-            else:
-                new_dt.append((
-                    dt[0],
-                    dt[1].replace('S', 'U').replace('|', '')))
+def convert_to_gigabytes(s):
+    try:
+        gigs = float(s)
+    except ValueError:
+        slow = s.lower()
+
+        units = slow[-1]
+        amount = slow[:-1]
+        if units == 'g':
+            gigs = float(amount)
+        elif units == 'm':
+            gigs = float(amount) / 1024
+        elif units == 'k':
+            gigs = float(amount) / 1024 ** 2
+        elif units == 'b':
+            gigs = float(amount) / 1024 ** 3
+        else:
+            raise ValueError(f'band unit in {s}')
+
+    return gigs
+
+
+def convert_to_bytes(s):
+    try:
+        bts = float(s)
+    except ValueError:
+        slow = s.lower()
+
+        units = slow[-1]
+        amount = slow[:-1]
+        if units == 'g':
+            bts = float(amount) * 1024 ** 3
+        elif units == 'm':
+            bts = float(amount) * 1024 ** 2
+        elif units == 'k':
+            bts = float(amount) * 1024
+        elif units == 'b':
+            bts = float(amount)
         else:
-            new_dt.append(dt)
+            raise ValueError(f'band unit in {s}')
+
+    return bts
+
+
+def get_compression_with_defaults(compression=None, convert=False):
+    """
+    get compression with defaults set
+
+    Parameters
+    ----------
+    compression: dict, optional
+        If a dict is sent, defaults are filled in as needed, otherwise defaults
+        are returned
+    convert: bool, optional
+        If set to True, convert the shuffle to the blosc integer
+        value
+
+    Returns
+    -------
+    dict with compression set
+    """
+
+    comp = defaults.DEFAULT_COMPRESSION.copy()
+    if hasattr(compression, 'keys'):
+        comp.update(compression)
+
+    if convert:
+        comp['shuffle'] = convert_shuffle(comp['shuffle'])
+
+    return comp
+
+
+def convert_shuffle(shuffle):
+    """
+    convert shuffle to the integer value
+
+    Parameters
+    ----------
+    shuffle: str or int
+        'shuffle', blosc.SHUFFLE, 'bitshuffle' or blosc.BITSHUFFLE
+        'noshuffle', blosc.NOSHUFFLE
+        String is case insensitive
+
+    Returns
+    -------
+    The integer value, e.g. blosc.SHUFFLE
+    """
+    import blosc
+
+    shuffle = shuffle.upper()
+
+    if shuffle in ('SHUFFLE', blosc.SHUFFLE):
+        new_shuf = blosc.SHUFFLE
+    elif shuffle in ('BITSHUFFLE', blosc.BITSHUFFLE):
+        new_shuf = blosc.BITSHUFFLE
+    elif shuffle in ('NOSHUFFLE', blosc.NOSHUFFLE):
+        new_shuf = blosc.NOSHUFFLE
+    else:
+        raise ValueError(f'bad shuffle: {shuffle}')
+    return new_shuf
+
+
+def schema_to_dtype(schema):
+    """
+    convert a schema into a numpy dtype
+
+    Parameters
+    ----------
+    schema: dict
+        Dict names for fields and dtype
+    """
+    descr = []
+
+    for name in schema:
+        descr.append((name, schema[name]['dtype']))
+
+    return np.dtype(descr)
+
+
+def get_chunks(chunkrows_sorted, rows):
+    """
+    Get chunk assignments for the input rows
+
+    Parameters
+    ----------
+    chunkrows_sorted: array
+        Sorted array of chunk row start positions
+    rows: array
+        Rows to assign
+
+    Returns
+    -------
+    chunk_indices: array
+        chunk index for each row
+    """
+    s = np.searchsorted(chunkrows_sorted, rows, side='right')
+    s -= 1
+    s.clip(min=0, max=chunkrows_sorted.size-1, out=s)
+    return s
+
+
+def get_data_names(data):
+    """
+    Get names for the data, either keys for a dict of arrays or names
+    from a structured array
+
+    Parameters
+    ----------
+    data: array or dict
+        A structured array or dict with arrays
+
+    Returns
+    -------
+    List of names
+    """
+    if hasattr(data, 'keys'):
+        names = list(data.keys())
+    else:
+        names = data.dtype.names
+        if names is None:
+            raise ValueError('array must have fields')
+
+    return names
+
+
+def byteswap_inplace(data):
+    """
+    Byte swap the data in place and with new dtype
+    """
+    data.byteswap(inplace=True)
+    data.dtype = data.dtype.newbyteorder()
+
+
+def get_data_with_conversion(data, dtype, ndmin=1):
+    """
+    returns the data, possibly converted to the specified type,
+    otherwise just a new ref
+    """
+    try:
+        ndata = np.array(data, ndmin=ndmin, dtype=dtype, copy=False)
+    except ValueError as err:
+        if isinstance(data, np.ndarray):
+            raise ValueError(
+                f'Could not convert data of type {data.dtype} to {dtype}: '
+                f'{err}'
+            )
+        else:
+            raise
+
+    return ndata
+
+
+def get_sub_dir(root, name):
+    if name is None:
+        return None
+
+    check_sub_name(name)
+
+    nsub = '.cols/'.join(name.split('/'))
+    return os.path.join(root, nsub)
+
+
+def check_sub_name(name):
+    if len(name) < 2 or name[-1] != '/' or name[0] == '/':
+        raise ValueError(
+            f'sub-Columns names must be a string with the form '
+            f'path/, got {name}'
+        )
+
+
+def get_sub_name(name):
+    """
+    sub-cols get a leading slash
+    """
+    return f'{name}/'
+
 
-    new_dtype = np.dtype(new_dt)
-    return new_dtype, do_conversion
+def iscols(arg):
+    return not isinstance(arg, slice) and isinstance(arg[0], str)
diff --git a/pycolumns/version.py b/pycolumns/version.py
index 1f356cc..99507ff 100644
--- a/pycolumns/version.py
+++ b/pycolumns/version.py
@@ -1,1 +1,1 @@
-__version__ = '1.0.0'
+__version__ = '2.0.0a'
diff --git a/setup.py b/setup.py
index 0e8a9d2..d1fa5c6 100644
--- a/setup.py
+++ b/setup.py
@@ -1,5 +1,7 @@
 import os
 from setuptools import setup, find_packages
+from setuptools.extension import Extension
+from setuptools.command import build_ext
 
 __version__ = None
 pth = os.path.join(
@@ -14,6 +16,22 @@
 with open(os.path.join(os.path.dirname(__file__), "README.md")) as fp:
     long_description = fp.read()
 
+ext = Extension(
+    "pycolumns._column_pywrap",
+    ["pycolumns/_column_pywrap.c"],
+)
+
+
+class BuildExt(build_ext.build_ext):
+    '''Custom build_ext command to hide the numpy import
+    Inspired by http://stackoverflow.com/a/21621689/1860757'''
+    def finalize_options(self):
+        '''add numpy includes to the include dirs'''
+        build_ext.build_ext.finalize_options(self)
+        import numpy as np
+        self.include_dirs.append(np.get_include())
+
+
 setup(
     name='pycolumns',
     packages=find_packages(),
@@ -23,6 +41,8 @@
     description='A simple, efficient, pythonic column data store',
     long_description=long_description,
     long_description_content_type='text/markdown; charset=UTF-8; variant=GFM',
-    setup_requires=['numpy', 'fitsio'],
-    install_requires=['numpy', 'fitsio'],
+    setup_requires=['numpy', 'blosc'],
+    install_requires=['numpy', 'blosc'],
+    ext_modules=[ext],
+    cmdclass={'build_ext': BuildExt},
 )


Test patch:
diff --git a/pycolumns/tests/test_access.py b/pycolumns/tests/test_access.py
new file mode 100644
index 0000000..793f3ae
--- /dev/null
+++ b/pycolumns/tests/test_access.py
@@ -0,0 +1,314 @@
+import pytest
+
+
+def get_data(rng):
+    import numpy as np
+
+    num = 20
+    dtype = [('id', 'i8'), ('rand', 'f4'), ('scol', 'U5')]
+    data = np.zeros(num, dtype=dtype)
+    data['id'] = np.arange(num)
+    data['rand'] = rng.uniform(size=num)
+    data['scol'] = [
+        's' + str(data['id'][i]) for i in range(num)
+    ]
+
+    subnum = 25
+    sub_data = np.zeros(subnum, dtype=[('ra', 'f8'), ('dec', 'f8')])
+    sub_data['ra'] = rng.uniform(size=sub_data.size)
+    sub_data['dec'] = rng.uniform(size=sub_data.size)
+
+    sub2num = 15
+    sub2_data = np.zeros(sub2num, dtype=[('x', 'f4')])
+    sub2_data['x'] = rng.uniform(size=sub2_data.size)
+
+    return data, sub_data, sub2_data
+
+
+@pytest.mark.parametrize('compression', [False, True])
+def test_access(compression):
+    """
+    cache_mem of 0.01 will force use of mergesort
+    """
+    import os
+    import tempfile
+    import numpy as np
+    from ..columns import Columns
+    from ..indices import Indices
+
+    seed = 333
+
+    rng = np.random.RandomState(seed)
+
+    data, sub_data, sub2_data = get_data(rng)
+
+    if compression:
+        ccols = ['id', 'scol']
+    else:
+        ccols = None
+
+    with tempfile.TemporaryDirectory() as tmpdir:
+
+        cdir = os.path.join(tmpdir, 'test.cols')
+
+        # Create a new Columns store with data in the root
+        cols = Columns.create_from_array(
+            cdir,
+            data=data,
+            compression=ccols,
+            verbose=True,
+        )
+
+        assert len(cols.names) == len(data.dtype.names)
+
+        versions = {'numpy': '1.23.5'}
+        cols.create_meta('versions', versions)
+
+        rmeta = cols.meta['versions'].read()
+        assert rmeta == versions
+
+        cols.create_meta('list', [3, 4])
+        assert cols.meta['list'].read() == [3, 4]
+
+        indata = cols.read()
+        for name in data.dtype.names:
+            assert np.all(data[name] == indata[name])
+
+        indata = cols[:]
+        for name in data.dtype.names:
+            assert np.all(data[name] == indata[name])
+
+        indata = cols[20:30]
+        for name in data.dtype.names:
+            assert np.all(data[name][20:30] == indata[name])
+
+        for name in data.dtype.names:
+            tdata = cols[name][:]
+            assert np.all(data[name] == tdata)
+
+        columns = ['id', 'rand']
+        indata = cols[columns][:]
+        for name in indata.dtype.names:
+            assert np.all(data[name] == indata[name])
+
+        rows = [8, 9, 15]
+        indata = cols.read(rows=rows, columns=columns)
+        for name in indata.dtype.names:
+            assert np.all(data[name][rows] == indata[name])
+
+        indata = cols[columns][rows]
+        for name in indata.dtype.names:
+            assert np.all(data[name][rows] == indata[name])
+
+        # sorted rows
+        rows = np.arange(0, data.size, 2)
+        indata = cols.read(rows=rows)
+        for name in data.dtype.names:
+            assert np.all(data[name][rows] == indata[name])
+
+        for name in data.dtype.names:
+            assert np.all(data[name][rows] == cols[name][rows])
+
+        for name in data.dtype.names:
+            assert np.all(data[name][10:15] == cols[name][10:15])
+            assert np.all(data[name][-5:-2] == cols[name][-5:-2])
+
+        for name in data.dtype.names:
+            assert np.all(data[name][5] == cols[name][5])
+
+        #
+        # update data in columns and meta
+        #
+
+        dadd = {'pycolumns': '2.0.0'}
+        cols.meta['versions'].update(dadd)
+        versions.update(dadd)
+
+        assert cols.meta['versions'].read() == versions
+
+        dnew = {'replaced': 3}
+        cols.meta['versions'].write(dnew)
+        assert cols.meta['versions'].read() == dnew
+
+        for name in data.dtype.names:
+            data[name][10:15] = (
+                np.arange(1000, 1000+5).astype(data[name].dtype)
+            )
+            cols[name][10:15] = data[name][10:15]
+            assert np.all(data[name][:] == cols[name][:])
+
+        # sorted rows
+        rows = [5, 15, 17]
+        vals = np.arange(len(rows))
+        for name in data.dtype.names:
+            data[name][rows] = vals.astype(data[name].dtype)
+            cols[name][rows] = data[name][rows]
+            assert np.all(data[name][:] == cols[name][:])
+
+        # unsorted rows
+        rows = [17, 5, 15]
+        vals = np.arange(len(rows))
+        for name in data.dtype.names:
+            data[name][rows] = vals.astype(data[name].dtype)
+            cols[name][rows] = data[name][rows]
+            assert np.all(data[name][:] == cols[name][:])
+
+        row = 8
+        val = 1999
+        for name in data.dtype.names:
+            data[name][row] = val
+            cols[name][row] = val
+            assert np.all(data[name][:] == cols[name][:])
+
+        # check negative indices
+        num = cols.size
+        assert cols[name][-2] == cols[name][num-2]
+        assert np.all(cols[name][[-3, -1]] == cols[name][[num-3, num-1]])
+        assert np.all(cols[name][-3:] == cols[name][[num-3, num-2, num-1]])
+
+        with pytest.raises(ValueError):
+            # tring to set number from string not representing numbers
+            cols['rand'][:] = cols['scol'][:]
+
+        # automatic conversion works num -> string
+        cols['scol'][:] = np.arange(cols.nrows)
+        assert np.all(
+            cols['scol'][:] == np.arange(cols.nrows).astype('U5')
+        )
+
+        cols['rand'][:] = cols['id'][:]
+        assert np.all(
+            cols['rand'][:] == cols['id'][:].astype(cols['rand'].dtype)
+        )
+
+        # filling slice with scalar
+        cols['id'][5:10] = 3
+        assert np.all(cols['id'][5:10] == 3)
+
+        # filling slice with scalar
+        cols['id'][-5:-2] = 9
+        assert np.all(cols['id'][-5:-2] == 9)
+
+        cols['scol'][5:10] = 'test'
+        assert np.all(cols['scol'][5:10] == 'test')
+
+        # filling rows with a scalar
+        ind = Indices([3, 5, 8], is_sorted=True)
+        cols['id'][ind] = 9999
+        assert np.all(cols['id'][ind] == 9999)
+
+        cols['scol'][ind] = 'test'
+        assert np.all(cols['scol'][ind] == 'test')
+
+        # filling rows unsorted
+        ind = [8, 1, 3]
+        cols['id'][ind] = -8888
+        assert np.all(cols['id'][ind] == -8888)
+
+        cols['scol'][ind] = '333'
+        assert np.all(cols['scol'][ind] == '333')
+
+        # This only works filling gall items or if data
+        # has the right length
+        cols['rand'] = 8
+        assert np.all(cols['rand'][:] == 8)
+
+        d = rng.uniform(size=cols.nrows).astype('f4')
+        cols['rand'] = d
+        assert np.all(cols['rand'][:] == d)
+
+        with pytest.raises(IndexError):
+            # not long enough
+            cols['rand'] = [3, 4]
+
+        #
+        # adding more tables
+        #
+
+        cols.from_array(name='sub1/', data=sub_data)
+        cols['sub1/']
+        cols['sub1/dec']
+
+        assert len(cols.names) == len(data.dtype.names) + 1
+
+        cols['sub1/'].from_array(name='sub2/', data=sub2_data)
+        cols['sub1/']['sub2/']
+        cols['sub1/sub2/']
+        cols['sub1/sub2/x']
+
+        # use of "in"
+        assert 'sub1/sub2/' in cols
+
+        cols.from_array(name='sub1/sub2/sub3/', data=sub2_data)
+        cols['sub1/sub2/sub3/']
+        cols['sub1/sub2/sub3/x']
+
+        cols2 = Columns(cdir)
+        cols2['sub1/']['dec']
+        cols2['sub1/dec']
+        cols2['sub1/sub2/']
+        cols2['sub1/sub2/x']
+        cols2['sub1/sub2/sub3/x']
+
+        with pytest.raises(IndexError):
+            cols['sub1/sub2/sub3/sub4/']
+
+        ra = cols['sub1/']['ra'][:]
+        assert np.all(ra == sub_data['ra'])
+        assert np.all(cols['sub1/ra'][:] == sub_data['ra'])
+        with pytest.raises(TypeError):
+            cols['sub1/'] = 5
+
+        x = cols['sub1/sub2/']['x'][:]
+        assert np.all(x == sub2_data['x'])
+        x = cols['sub1/sub2/x'][:]
+        assert np.all(x == sub2_data['x'])
+
+
+def test_set_compressed():
+    """
+    cache_mem of 0.01 will force use of mergesort
+    """
+    import os
+    import tempfile
+    import numpy as np
+    from ..columns import Columns
+
+    seed = 333
+    num = 100_000
+    chunksize = '10000r'
+
+    rng = np.random.RandomState(seed)
+
+    dtype = [('id', 'i8'), ('rand', 'f4'), ('scol', 'U5')]
+    data = np.zeros(num, dtype=dtype)
+    data['id'] = np.arange(num)
+    data['rand'] = rng.uniform(size=num)
+    data['scol'] = [
+        's' + str(data['id'][i]) for i in range(num)
+    ]
+
+    ccols = ['id', 'scol']
+
+    with tempfile.TemporaryDirectory() as tmpdir:
+
+        cdir = os.path.join(tmpdir, 'test.cols')
+        cols = Columns.create(cdir, verbose=True)
+        cols.from_array(data=data, compression=ccols, chunksize=chunksize)
+
+        assert np.all(cols['id'][:] == data['id'])
+
+        ndata = rng.randint(0, 2**16, size=cols.size)
+        cols['id'][:] = ndata
+        assert np.all(cols['id'][:] == ndata)
+
+        cols['id'][:] = 999
+        assert np.all(cols['id'][:] == 999)
+
+        print(f"nchunks: {cols['id']._col.nchunks}")
+        cols.append(data)
+        print(f"nchunks: {cols['id']._col.nchunks}")
+        assert np.all(cols['id'][data.size:] == data['id'])
+
+        ndata = rng.randint(0, 2**16, size=cols.size)
+        cols['id'][data.size:] = ndata
diff --git a/pycolumns/tests/test_ccolumn.py b/pycolumns/tests/test_ccolumn.py
new file mode 100644
index 0000000..ed8a211
--- /dev/null
+++ b/pycolumns/tests/test_ccolumn.py
@@ -0,0 +1,145 @@
+import pytest
+
+
+@pytest.mark.parametrize('dtype', ['i8', 'f4', 'U3'])
+def test_column(dtype):
+    import os
+    import tempfile
+    import numpy as np
+    from .._column import Column
+
+    seed = 333
+    num = 20
+
+    rng = np.random.RandomState(seed)
+
+    if dtype[0] == 'i':
+        data = np.arange(num, dtype=dtype)
+    elif dtype[0] == 'U':
+        data = np.zeros(num, dtype=dtype)
+        data[:] = [str(i) for i in range(num)]
+    else:
+        data = rng.uniform(size=num).astype(dtype)
+
+    with tempfile.TemporaryDirectory() as tmpdir:
+        fname = os.path.join(tmpdir, 'test.col')
+        with Column(fname, dtype=data.dtype, mode='w+', verbose=True) as col:
+            print('-' * 70)
+            print('before append')
+            print(col)
+
+            col.append(data)
+
+            print('-' * 70)
+            print('before append')
+            print(col)
+
+            indata = col[:]
+            assert np.all(indata == data)
+
+            indata = col[2:8]
+            assert np.all(indata == data[2:8])
+
+            indata = col[2:18:2]
+            assert np.all(indata == data[2:18:2])
+
+            ind = [3, 5, 7]
+            indata = col[ind]
+            assert np.all(indata == data[ind])
+
+            # out of order
+            ind = [7, 3, 5]
+            indata = col[ind]
+            assert np.all(indata == data[ind])
+
+            ind = 5
+            indata = col[ind]
+            assert np.all(indata == data[ind])
+
+            s = slice(2, 8)
+            indata = np.zeros(s.stop - s.start, dtype=data.dtype)
+            col.read_slice_into(indata, s)
+            assert np.all(indata == data[s])
+
+            s = slice(-3, -1)
+            indata = np.zeros(2, dtype=data.dtype)
+            col.read_slice_into(indata, s)
+            assert np.all(indata == data[s])
+
+            ind = [3, 5, 7]
+            indata = np.zeros(len(ind), dtype=data.dtype)
+            col.read_rows_into(indata, ind)
+            assert np.all(indata == data[ind])
+
+            ind = 6
+            indata = np.zeros(1, dtype=data.dtype)
+            col.read_row_into(indata, ind)
+            assert np.all(indata[0] == data[ind])
+
+            # cannot read into array scalar
+            with pytest.raises(ValueError):
+                ind = 6
+                indata = np.zeros(1, dtype=data.dtype)
+                col.read_row_into(indata[0], ind)
+
+            # mismatch in ind size and data size
+            with pytest.raises(ValueError):
+                ind = [3, 5, 7]
+                indata = np.zeros(5, dtype=data.dtype)
+                col.read_rows_into(indata, ind)
+
+            # mismatch in ind size and data size
+            with pytest.raises(ValueError):
+                ind = 3
+                indata = np.zeros(5, dtype=data.dtype)
+                col.read_row_into(indata, ind)
+
+            col.append(data)
+            assert col.nrows == data.size * 2
+
+            #
+            # updating column data
+            #
+
+            one = data[:1]
+            row2update = 3
+            col.update_row(row2update, one)
+            check = col[3]
+            assert check == one[0]
+
+            two = data[3:3+2]
+            start = 15
+            col.write_at(two, start)
+            check = col[start:start+2]
+            assert np.all(check == two)
+
+            col[5] = data[0]
+            assert np.all(col[5] == data[0])
+
+            col[:data.size] = data
+            assert np.all(col[:data.size] == data)
+
+            # sorted rows
+            rows = [0, 2, 5]
+            wdata = np.array([99, 88, 77], dtype=dtype)
+            col[rows] = wdata
+            assert np.all(col[rows] == wdata)
+
+            # out of order rows
+            rows = [8, 3, 7]
+            wdata = np.array([100, 102, 103], dtype=dtype)
+            col[rows] = wdata
+            assert np.all(col[rows] == wdata)
+
+            rows = slice(0, 6, 2)
+            wdata = np.arange(0, 6, 2).astype(dtype)
+            col[rows] = wdata
+            assert np.all(col[rows] == wdata)
+
+            # bad slice size
+            with pytest.raises(IndexError):
+                col[0:5] = wdata
+
+            # out of bounds update
+            with pytest.raises(IndexError):
+                col[np.arange(100)] = np.arange(100)
diff --git a/pycolumns/tests/test_chunks.py b/pycolumns/tests/test_chunks.py
new file mode 100644
index 0000000..220bcb7
--- /dev/null
+++ b/pycolumns/tests/test_chunks.py
@@ -0,0 +1,84 @@
+import pytest
+
+
+@pytest.mark.parametrize('compression', [False, True])
+@pytest.mark.parametrize('dtype', ['i8', 'f4', 'U3'])
+def test_chunks(compression, dtype):
+    import os
+    import tempfile
+    import numpy as np
+    from ..chunks import Chunks
+
+    seed = 812
+    num = 100
+
+    rng = np.random.RandomState(seed)
+
+    if dtype[0] == 'i':
+        data = np.arange(num, dtype=dtype)
+    elif dtype[0] == 'U':
+        data = np.zeros(num, dtype=dtype)
+        data[:] = [str(i) for i in range(num)]
+    else:
+        data = rng.uniform(size=num).astype(dtype)
+
+    # small chunksize to guarantee we cross chunks
+    row_chunksize = 10
+    chunksize = f'{row_chunksize}r'
+
+    with tempfile.TemporaryDirectory() as tmpdir:
+        fname = os.path.join(tmpdir, 'test.array')
+        chunks_fname = os.path.join(tmpdir, 'test.chunks')
+
+        with open(fname, 'w') as fobj:  # noqa
+            pass
+        with open(chunks_fname, 'w') as fobj:  # noqa
+            pass
+
+        with Chunks(
+            fname,
+            chunks_fname,
+            dtype=data.dtype,
+            chunksize=chunksize,
+            compression=compression,
+            mode='w+',
+            verbose=True,
+        ) as col:
+
+            print('-' * 70)
+            print('before append')
+            print(col)
+
+            col.append(data)
+            assert col.nchunks > 1
+            assert col.row_chunksize == row_chunksize
+
+            print('-' * 70)
+            print('before append')
+            print(col)
+
+            indata = col[:]
+            assert np.all(indata == data)
+
+            indata = col[2:8]
+            assert np.all(indata == data[2:8])
+
+            indata = col[2:18:2]
+            assert np.all(indata == data[2:18:2])
+
+            ind = [3, 5, 7]
+            indata = col[ind]
+            assert np.all(indata == data[ind])
+
+            # out of order
+            ind = np.arange(num)
+            rng.shuffle(ind)
+            indata = col[ind]
+            assert np.all(indata == data[ind])
+
+            ind = 5
+            indata = col[ind]
+            assert np.all(indata == data[ind])
+
+            col.append(data)
+            assert col.nrows == data.size * 2
diff --git a/pycolumns/tests/test_create.py b/pycolumns/tests/test_create.py
index baf7af7..74b59a0 100644
--- a/pycolumns/tests/test_create.py
+++ b/pycolumns/tests/test_create.py
@@ -1,9 +1,12 @@
 import pytest
 
 
-@pytest.mark.parametrize('cache_mem', [1.0, 0.01])
+@pytest.mark.parametrize('cache_mem', ['1g', '10k'])
+@pytest.mark.parametrize('compression', [False, True])
 @pytest.mark.parametrize('verbose', [True, False])
-def test_create(cache_mem, verbose):
+@pytest.mark.parametrize('fromdict', [False, True])
+@pytest.mark.parametrize('from_array', [False, True])
+def test_create(cache_mem, compression, verbose, fromdict, from_array):
     """
     cache_mem of 0.01 will force use of mergesort
     """
@@ -11,36 +14,68 @@ def test_create(cache_mem, verbose):
     import tempfile
     import numpy as np
     from ..columns import Columns
+    from ..schema import TableSchema
 
     seed = 333
     num = 20
 
     rng = np.random.RandomState(seed)
 
+    dtype = [('id', 'i8'), ('rand', 'f4'), ('scol', 'U5')]
+    data = np.zeros(num, dtype=dtype)
+    data['id'] = np.arange(num)
+    data['rand'] = rng.uniform(size=num)
+    data['scol'] = [str(val) for val in data['id']]
+
+    if compression:
+        ccols = ['id', 'scol']
+    else:
+        ccols = None
+
+    if fromdict:
+        ddict = {}
+        for name in data.dtype.names:
+            ddict[name] = data[name]
+        schema = TableSchema.from_array(ddict, compression=ccols)
+    else:
+        schema = TableSchema.from_array(data, compression=ccols)
+
+    print(schema)
+
     with tempfile.TemporaryDirectory() as tmpdir:
+
         cdir = os.path.join(tmpdir, 'test.cols')
-        cols = Columns(cdir, cache_mem=cache_mem, verbose=verbose)
-        assert len(cols.names) == 0
+        cols = Columns.create(cdir, cache_mem=cache_mem, verbose=verbose)
 
         assert cols.dir == cdir
         assert cols.verbose == verbose
         assert cols.cache_mem == cache_mem
 
-        dtype = [('id', 'i8'), ('rand', 'f4'), ('scol', 'U5')]
-        data = np.zeros(num, dtype=dtype)
-        data['id'] = np.arange(num)
-        data['rand'] = rng.uniform(size=num)
-        data['scol'] = [str(val) for val in data['id']]
+        if fromdict:
+            append_data = ddict
+        else:
+            append_data = data
 
-        cols.append(data)
+        # created in root
+        if from_array:
+            cols.from_array(data=append_data, compression=ccols)
+        else:
+            cols.create_table(schema=schema)
+            cols.append(append_data)
 
         assert len(cols.names) == len(data.dtype.names)
         meta = {'version': '0.1', 'seeing': 0.9}
-        cols.create_column('meta', 'dict')
-        cols['meta'].write(meta)
+        cols.create_meta('metadata', meta)
+
+        assert cols.meta['metadata'].read() == meta
 
-        rmeta = cols['meta'].read()
-        assert rmeta == meta
+        newd = {'x': 5, 'name': 'joe'}
+        meta.update(newd)
+        cols.meta['metadata'].update(newd)
+        assert cols.meta['metadata'].read() == meta
+
+        ncols = Columns(cdir)
+        assert ncols.meta['metadata'].read() == meta
 
         # after appending, verify is run, but let's double check
         for name in data.dtype.names:
@@ -74,11 +109,68 @@ def test_create(cache_mem, verbose):
             )
             cols.append(bad_data)
 
-        # can currently update column at a time
-        # cols['rand'][5] = 35
-        # assert cols['rand'][5] == 35
 
-        # idx = [8, 12]
-        # vals = [1.0, 2.0]
-        # cols['rand'][idx] = vals
-        # assert np.all(cols['rand'][idx] == vals)
+def test_create_column():
+    """
+    cache_mem of 0.01 will force use of mergesort
+    """
+    import os
+    import tempfile
+    import numpy as np
+    from ..columns import Columns
+    from ..schema import TableSchema, ColumnSchema
+
+    seed = 333
+    num = 20
+
+    rng = np.random.RandomState(seed)
+
+    dtype = [('id', 'i8'), ('rand', 'f4'), ('scol', 'U5')]
+    data = np.zeros(num, dtype=dtype)
+    data['id'] = np.arange(num)
+    data['rand'] = rng.uniform(size=num)
+    data['scol'] = [str(val) for val in data['id']]
+
+    ccols = ['id', 'scol']
+
+    schema = TableSchema.from_array(data, compression=ccols)
+
+    print(schema)
+
+    with tempfile.TemporaryDirectory() as tmpdir:
+
+        cdir = os.path.join(tmpdir, 'test.cols')
+        cols = Columns.create_from_array(
+            cdir,
+            data,
+            compression=ccols,
+            verbose=True,
+        )
+
+        newschema = ColumnSchema('newcol', dtype='i4')
+        cols.create_column(newschema)
+        assert cols['newcol'][:].size == cols.size
+        assert np.all(cols['newcol'][:] == 0)
+
+        newschema = ColumnSchema('news', dtype='U2')
+        cols.create_column(newschema)
+        assert cols['news'][:].size == cols.size
+        assert np.all(cols['news'][:] == '')
+
+        newschema = ColumnSchema('fcol', dtype='U2', fill_value='-')
+        cols.create_column(newschema)
+        assert np.all(cols['fcol'][:] == '-')
+
+        newschema = ColumnSchema('x9', dtype='f4', fill_value=9.5)
+        cols.create_column(newschema)
+        assert np.all(cols['x9'][:] == 9.5)
+
+        newschema = ColumnSchema('ss', dtype='S3', fill_value='yes')
+        cols.create_column(newschema)
+        assert np.all(cols['ss'][:] == b'yes')
+
+        newschema = ColumnSchema(
+            'scomp', dtype='U2', compression=True, fill_value='hi',
+        )
+        cols.create_column(newschema)
+        assert np.all(cols['scomp'][:] == 'hi')
diff --git a/pycolumns/tests/test_create_index.py b/pycolumns/tests/test_create_index.py
index 8e64cd3..6bb70e1 100644
--- a/pycolumns/tests/test_create_index.py
+++ b/pycolumns/tests/test_create_index.py
@@ -2,70 +2,150 @@
 
 
 @pytest.mark.parametrize('cache_mem', [1.0, 0.01])
-def test_create_index(cache_mem):
+@pytest.mark.parametrize('compression', [False, True])
+def test_create_index(cache_mem, compression):
     """
     cache_mem of 0.01 will force use of mergesort
     """
     import os
     import tempfile
     import numpy as np
-    import fitsio
+    from .. import _column
     from ..columns import Columns
 
     seed = 333
     num = 1_000_000
+    rng = np.random.RandomState(seed)
+    data = np.zeros(num, dtype=[('rand', 'f8')])
+    data['rand'] = rng.uniform(size=num)
+
+    if compression:
+        ccols = ['id', 'scol']
+    else:
+        ccols = None
 
     with tempfile.TemporaryDirectory() as tmpdir:
-        cdir = os.path.join(tmpdir, 'test.cols')
-        cols = Columns(cdir, cache_mem=cache_mem, verbose=True)
 
-        rng = np.random.RandomState(seed)
-        data = np.zeros(num, dtype=[('rand', 'f8')])
-        data['rand'] = rng.uniform(size=num)
+        cdir = os.path.join(tmpdir, 'test.cols')
+        cols = Columns.create_from_array(
+            cdir,
+            data=data,
+            compression=ccols,
+            cache_mem=cache_mem,
+            verbose=True,
+        )
 
-        cols.append(data)
         cols['rand'].create_index()
         assert cols['rand'].has_index
 
         ifile = cols['rand'].index_filename
-        idata = fitsio.read(ifile)
+        sfile = cols['rand'].sorted_filename
+        idata = _column.read(ifile, dtype=cols['rand'].index_dtype)
+        sdata = _column.read(sfile, dtype=cols['rand'].dtype)
 
         s = data['rand'].argsort()
-        assert np.all(idata['value'] == data['rand'][s])
+        assert np.all(idata == s)
+        assert np.all(sdata == data['rand'][s])
 
 
-def test_create_index_str():
+@pytest.mark.parametrize('compression', [False, True])
+def test_create_index_str(compression):
     """
     cache_mem of 0.01 will force use of mergesort
     """
     import os
     import tempfile
     import numpy as np
-    import fitsio
+    from .. import _column
     from ..columns import Columns
 
     seed = 55
     num = 20
+    rng = np.random.RandomState(seed)
+
+    dt = 'U5'
+    data = np.zeros(num, dtype=[('scol', dt)])
+    rand = rng.uniform(size=num)
+    data['scol'] = rand.astype(dt)
+
+    if compression:
+        ccols = ['id', 'scol']
+    else:
+        ccols = None
 
     with tempfile.TemporaryDirectory() as tmpdir:
-        cdir = os.path.join(tmpdir, 'test.cols')
-        cols = Columns(cdir, verbose=True)
 
-        rng = np.random.RandomState(seed)
-        data = np.zeros(num, dtype=[('scol', 'U5')])
-        rand = rng.uniform(size=num)
-        data['scol'] = ['%.2g' % v for v in rand]
+        cdir = os.path.join(tmpdir, 'test.cols')
+        cols = Columns.create_from_array(
+            cdir,
+            data=data,
+            compression=ccols,
+            verbose=True,
+        )
 
-        cols.append(data)
         cols['scol'].create_index()
         assert cols['scol'].has_index
 
         ifile = cols['scol'].index_filename
-        idata = fitsio.read(ifile)
+        sfile = cols['scol'].sorted_filename
+        idata = _column.read(ifile, dtype=cols['scol'].index_dtype)
+        sdata = _column.read(sfile, dtype=cols['scol'].dtype)
 
         s = data['scol'].argsort()
-        assert np.all(idata['value'] == data['scol'][s])
+        assert np.all(idata == s)
+        assert np.all(sdata == data['scol'][s])
 
         ind = (data['scol'] == data['scol'][5])
         rd = data['scol'][ind]
         assert rd.size == 1
+
+
+def test_updating():
+    """
+    cache_mem of 0.01 will force use of mergesort
+    """
+    import os
+    import tempfile
+    import numpy as np
+    from ..columns import Columns
+
+    num = 20
+    data = np.zeros(num, dtype=[('ind', 'i8')])
+    data['ind'] = np.arange(num)
+
+    with tempfile.TemporaryDirectory() as tmpdir:
+
+        cdir = os.path.join(tmpdir, 'test.cols')
+
+        cols = Columns.create_from_array(
+            cdir,
+            data=data,
+            verbose=True,
+        )
+
+        cols['ind'].create_index()
+        assert cols['ind'].has_index
+        assert cols['ind']._index.size == cols['ind'].size
+
+        # in context, indexes are not updated during an append
+        with cols.updating():
+            cols.append(data)
+
+            assert cols['ind']._index.size != cols['ind'].size
+            assert cols['ind'].size == 2 * data.size
+            assert cols['ind']._index.size == data.size
+
+        assert cols['ind']._index.size == cols['ind'].size
+
+        cols['ind'][:] = np.arange(cols.nrows)
+        with cols['ind'].updating():
+            cols['ind'][5:10] = 88
+            assert np.all(cols['ind'][5:10] == 88)
+
+            # should not return any
+            w = cols['ind'] == 88
+            assert w.size == 0
+
+        # should not return any
+        w = cols['ind'] == 88
+        assert w.size == 5
diff --git a/pycolumns/tests/test_delete.py b/pycolumns/tests/test_delete.py
new file mode 100644
index 0000000..b662834
--- /dev/null
+++ b/pycolumns/tests/test_delete.py
@@ -0,0 +1,57 @@
+# import pytest
+
+
+def test_delete():
+    import os
+    import tempfile
+    import numpy as np
+    from ..columns import Columns
+
+    seed = 812
+    num = 20
+
+    rng = np.random.RandomState(seed)
+
+    dtype = [('id', 'i8'), ('rand', 'f4'), ('scol', 'U5')]
+    data = np.zeros(num, dtype=dtype)
+    data['id'] = np.arange(num)
+    data['rand'] = rng.uniform(size=num)
+    data['scol'] = [str(val) for val in data['id']]
+
+    sub_data = np.zeros(num, dtype=[('ra', 'f8'), ('dec', 'f8')])
+    sub_data['ra'] = rng.uniform(size=sub_data.size)
+    sub_data['dec'] = rng.uniform(size=sub_data.size)
+
+    ccols = ['id', 'scol']
+
+    with tempfile.TemporaryDirectory() as tmpdir:
+
+        cdir = os.path.join(tmpdir, 'test.cols')
+
+        cols = Columns.create_from_array(
+            cdir,
+            data=data,
+            compression=ccols,
+            verbose=True,
+        )
+
+        cols.create_meta('metadata')
+        cols.meta['metadata'].write({'x': 3})
+        # cols['metadata'] = {'x': 3}
+
+        cols.from_array(name='sub/', data=sub_data)
+
+        iddir = cols['id'].dir
+        assert os.path.exists(iddir)
+
+        cols.delete_entry('id', yes=True)
+        assert not os.path.exists(iddir)
+
+        fname = cols.meta['metadata'].filename
+        assert os.path.exists(fname)
+
+        cols.delete_meta('metadata', yes=True)
+        assert not os.path.exists(fname)
+
+        cols.delete(yes=True)
+        assert not os.path.exists(cdir)
diff --git a/pycolumns/tests/test_query.py b/pycolumns/tests/test_query.py
index 1ba0331..a5e7046 100644
--- a/pycolumns/tests/test_query.py
+++ b/pycolumns/tests/test_query.py
@@ -15,15 +15,14 @@ def test_query():
 
     rng = np.random.RandomState(seed)
 
-    with tempfile.TemporaryDirectory() as tmpdir:
-        cdir = os.path.join(tmpdir, 'test.cols')
-        cols = Columns(cdir, verbose=True)
+    data = np.zeros(num, dtype=[('id', 'i8'), ('rand', 'f4')])
+    data['id'] = np.arange(num)
+    data['rand'] = rng.uniform(size=num, high=num)
 
-        data = np.zeros(num, dtype=[('id', 'i8'), ('rand', 'f4')])
-        data['id'] = np.arange(num)
-        data['rand'] = rng.uniform(size=num, high=num)
+    with tempfile.TemporaryDirectory() as tmpdir:
 
-        cols.append(data)
+        cdir = os.path.join(tmpdir, 'test.cols')
+        cols = Columns.create_from_array(cdir, data=data, verbose=True)
 
         for name in data.dtype.names:
             assert not cols[name].has_index
diff --git a/pycolumns/tests/test_schema.py b/pycolumns/tests/test_schema.py
new file mode 100644
index 0000000..2c9865c
--- /dev/null
+++ b/pycolumns/tests/test_schema.py
@@ -0,0 +1,180 @@
+import pytest
+
+
+def test_column_schema():
+    import numpy as np
+    from ..schema import ColumnSchema
+    from ..defaults import DEFAULT_COMPRESSION, DEFAULT_CHUNKSIZE
+
+    # simple
+    s = {'dtype': np.dtype('f8').str}
+    cs = ColumnSchema('blah', dtype='f8')
+    assert cs.name == 'blah'
+    assert cs == s
+
+    # with default compression
+    s = {
+        'dtype': np.dtype('f8').str,
+        'compression': DEFAULT_COMPRESSION.copy(),
+        'chunksize': DEFAULT_CHUNKSIZE,
+    }
+    cs = ColumnSchema('blah', dtype='f8', compression=True)
+    assert cs.name == 'blah'
+    assert cs == s
+
+    # with more specific settings
+    s = {
+        'dtype': np.dtype('f8').str,
+        'compression': {'cname': 'zlib'},
+        'chunksize': '1g',
+    }
+    cs = ColumnSchema(
+        'blah',
+        dtype='f8',
+        compression=s['compression'],
+        chunksize=s['chunksize'],
+    )
+    se = s.copy()
+    se['compression'] = DEFAULT_COMPRESSION.copy()
+    se['compression'].update(s['compression'])
+    assert cs.name == 'blah'
+    assert cs == se
+
+
+def test_table_schema_from_column_schema():
+    from ..schema import ColumnSchema, TableSchema
+    from ..defaults import DEFAULT_COMPRESSION, DEFAULT_CHUNKSIZE
+
+    cslist = [
+        ColumnSchema(name='id', dtype='i8', compression=True),
+        ColumnSchema(name='ra', dtype='f8'),
+        ColumnSchema(name='name', dtype='U5',
+                     compression={'shuffle': 'shuffle'}),
+    ]
+    ts = TableSchema(cslist)
+
+    for cs in cslist:
+        assert cs.name in ts
+        assert ts[cs.name] == cs
+
+    assert ts['id']['compression'] == DEFAULT_COMPRESSION
+    assert ts['id']['chunksize'] == DEFAULT_CHUNKSIZE
+    assert ts['name']['compression']['shuffle'] == 'shuffle'
+
+
+def test_table_schema_from_array():
+    import numpy as np
+    from ..schema import ColumnSchema, TableSchema
+    from ..defaults import DEFAULT_CHUNKSIZE
+
+    arr = np.zeros(1, dtype=[('id', 'i8'), ('ra', 'f8'), ('name', 'U5')])
+
+    # simplest
+    ts = TableSchema.from_array(arr)
+
+    cslist = []
+    for name in arr.dtype.names:
+        cs = ColumnSchema(name=name, dtype=arr[name].dtype)
+        cslist.append(cs)
+
+    for cs in cslist:
+        assert cs.name in ts
+        assert ts[cs.name] == cs
+
+    # some defaults
+    compression = ['id', 'name']
+    ts = TableSchema.from_array(arr, compression=compression)
+
+    cslist = []
+    for name in arr.dtype.names:
+        cs = ColumnSchema(
+            name=name, dtype=arr[name].dtype,
+            compression=name in compression,
+        )
+        cslist.append(cs)
+
+    for cs in cslist:
+        assert cs.name in ts
+        assert ts[cs.name] == cs
+
+    # get specific
+    compression = {
+        'id': {
+            'cname': 'zlib',
+            'clevel': 9,
+            'shuffle': 'shuffle',
+        },
+    }
+    chunksize = '1g'
+    ts = TableSchema.from_array(arr, compression=compression, chunksize='1g')
+
+    cslist = []
+    for name in arr.dtype.names:
+        cs = ColumnSchema(
+            name=name, dtype=arr[name].dtype,
+            compression=compression.get(name),
+            chunksize=chunksize,
+        )
+        cslist.append(cs)
+
+    for cs in cslist:
+        assert cs.name in ts
+        assert ts[cs.name] == cs
+
+    # get specific with chunksize too
+    compression = {
+        'id': {
+            'cname': 'zlib',
+            'clevel': 9,
+            'shuffle': 'shuffle',
+        },
+    }
+    chunksize = {'id': '1g'}
+
+    ts = TableSchema.from_array(
+        arr, compression=compression, chunksize=chunksize,
+    )
+
+    cslist = []
+    for name in arr.dtype.names:
+        cs = ColumnSchema(
+            name=name, dtype=arr[name].dtype,
+            compression=compression.get(name),
+            chunksize=chunksize.get(name, DEFAULT_CHUNKSIZE),
+        )
+        cslist.append(cs)
+
+    for cs in cslist:
+        assert cs.name in ts
+        assert ts[cs.name] == cs
+
+
+def test_column_compression():
+    from ..schema import _get_column_compression
+
+    assert _get_column_compression(False, 'blah') is None
+    assert _get_column_compression(True, 'blah') is True
+
+    assert _get_column_compression(['blah'], 'blah') is True
+    assert _get_column_compression(['hey'], 'blah') is None
+
+    assert _get_column_compression({'blah': True}, 'blah') is True
+    assert _get_column_compression({'blah': False}, 'blah') is None
+    assert _get_column_compression({'id': True}, 'blah') is None
+
+    comp = {'cname': 'zlib'}
+    assert _get_column_compression({'blah': comp}, 'blah') == comp
+
+
+def test_column_chunksize():
+    from ..schema import _get_column_chunksize
+    from ..defaults import DEFAULT_CHUNKSIZE
+
+    assert _get_column_chunksize('1m', 'blah') == '1m'
+    assert _get_column_chunksize(3, 'blah') == 3
+    assert _get_column_chunksize({'blah': '1g'}, 'blah') == '1g'
+    assert _get_column_chunksize({'id': '1m'}, 'blah') == DEFAULT_CHUNKSIZE
+    with pytest.raises(ValueError):
+        _get_column_chunksize({'id', '1m'}, 'blah')
+        _get_column_chunksize(['blah', 'id'], 'blah')
+        _get_column_chunksize(('blah', 'id'), 'blah')
diff --git a/pycolumns/tests/test_vacuum.py b/pycolumns/tests/test_vacuum.py
new file mode 100644
index 0000000..e2b8046
--- /dev/null
+++ b/pycolumns/tests/test_vacuum.py
@@ -0,0 +1,89 @@
+
+
+def test_vacuum():
+    import os
+    import tempfile
+    import numpy as np
+    from ..columns import Columns
+
+    seed = 9911
+    num = 200_000
+
+    rng = np.random.RandomState(seed)
+
+    dtype = [('id', 'i8')]
+    data = np.zeros(num, dtype=dtype)
+
+    # this will compress a lot
+    data['id'] = np.zeros(num)
+
+    with tempfile.TemporaryDirectory() as tmpdir:
+
+        cdir = os.path.join(tmpdir, 'test.cols')
+        cols = Columns.create(cdir, verbose=True)
+        cols.from_array(data, compression=['id'])
+
+        # this will expand quite a bit, fragmenting the file
+        rind = rng.randint(0, 2**31, size=num)
+        cols['id'][:] = rind
+        assert np.all(cols['id'][:] == rind)
+
+        cc = cols['id']._col
+        w, = np.where(cc.chunk_data['is_external'])
+        assert w.size > 0
+
+        for chunk_index in w:
+            ext_fname = cc.get_external_filename(chunk_index)
+            assert os.path.exists(ext_fname)
+
+        cols.vacuum()
+
+        assert np.all(cols['id'][:] == rind)
+
+        for chunk_index in w:
+            ext_fname = cc.get_external_filename(chunk_index)
+            assert not os.path.exists(ext_fname)
+
+
+def test_vacuum_context():
+    import os
+    import tempfile
+    import numpy as np
+    from ..columns import Columns
+
+    seed = 9911
+    num = 200_000
+
+    rng = np.random.RandomState(seed)
+
+    dtype = [('id', 'i8')]
+    data = np.zeros(num, dtype=dtype)
+
+    # this will compress a lot
+    data['id'] = np.zeros(num)
+
+    with tempfile.TemporaryDirectory() as tmpdir:
+
+        cdir = os.path.join(tmpdir, 'test.cols')
+        cols = Columns.create(cdir, verbose=True)
+        cols.from_array(data, compression=['id'])
+
+        # this will expand quite a bit, fragmenting the file
+        with cols['id'].updating(vacuum=True):
+            rind = rng.randint(0, 2**31, size=num)
+            cols['id'][:] = rind
+            assert np.all(cols['id'][:] == rind)
+
+            cc = cols['id']._col
+            w, = np.where(cc.chunk_data['is_external'])
+            assert w.size > 0
+
+            for chunk_index in w:
+                ext_fname = cc.get_external_filename(chunk_index)
+                assert os.path.exists(ext_fname)
+
+        assert np.all(cols['id'][:] == rind)
+
+        for chunk_index in w:
+            ext_fname = cc.get_external_filename(chunk_index)
+            assert not os.path.exists(ext_fname)