'Internal implementation of write.
`iterator` should be an iterator yielding pairs of (asset, ctable).'

def _write_internal(self, iterator, assets):

    total_rows = 0
    first_row = {}
    last_row = {}
    calendar_offset = {}
    columns = {k: carray(array([], dtype=uint32)) for k in US_EQUITY_PRICING_BCOLZ_COLUMNS}
    earliest_date = None
    sessions = self._calendar.sessions_in_range(self._start_session, self._end_session)
    if (assets is not None):
        @apply
        def iterator(iterator=iterator, assets=set(assets)):
            for (asset_id, table) in iterator:
                if (asset_id not in assets):
                    raise ValueError(('unknown    asset    id    %r' % asset_id))
                (yield (asset_id, table))
    for (asset_id, table) in iterator:
        nrows = len(table)
        for column_name in columns:
            if (column_name == 'id'):
                columns['id'].append(full((nrows,), asset_id, dtype='uint32'))
                continue
            columns[column_name].append(table[column_name])
        if (earliest_date is None):
            earliest_date = table['day'][0]
        else:
            earliest_date = min(earliest_date, table['day'][0])
        asset_key = str(asset_id)
        first_row[asset_key] = total_rows
        last_row[asset_key] = ((total_rows + nrows) - 1)
        total_rows += nrows
        table_day_to_session = compose(self._calendar.minute_to_session_label, partial(Timestamp, unit='s', tz='UTC'))
        asset_first_day = table_day_to_session(table['day'][0])
        asset_last_day = table_day_to_session(table['day'][(-1)])
        asset_sessions = sessions[sessions.slice_indexer(asset_first_day, asset_last_day)]
        assert (len(table) == len(asset_sessions)), 'Got    {}    rows    for    daily    bars    table    with    first    day={},    last    day={},    expected    {}    rows.\nMissing    sessions:    {}\nExtra    sessions:    {}'.format(len(table), asset_first_day.date(), asset_last_day.date(), len(asset_sessions), asset_sessions.difference(to_datetime(np.array(table['day']), unit='s', utc=True)).tolist(), to_datetime(np.array(table['day']), unit='s', utc=True).difference(asset_sessions).tolist())
        calendar_offset[asset_key] = sessions.get_loc(asset_first_day)
    full_table = ctable(columns=[columns[colname] for colname in US_EQUITY_PRICING_BCOLZ_COLUMNS], names=US_EQUITY_PRICING_BCOLZ_COLUMNS, rootdir=self._filename, mode='w')
    full_table.attrs['first_trading_day'] = (earliest_date if (earliest_date is not None) else iNaT)
    full_table.attrs['first_row'] = first_row
    full_table.attrs['last_row'] = last_row
    full_table.attrs['calendar_offset'] = calendar_offset
    full_table.attrs['calendar_name'] = self._calendar.name
    full_table.attrs['start_session_ns'] = self._start_session.value
    full_table.attrs['end_session_ns'] = self._end_session.value
    full_table.flush()
    return full_table
