IceDB v2 🧊 - An in-process Parquet merge engine to build dirt-cheap OLAP

What is IceDB?

The benefits of IceDB

Serverless

Powered by DuckDB

Based on Parquet

Why not other solutions?

ClickHouse

Spark, Flink, Hadoop, etc.

Presto and Trino

Managed Solutions like BigQuery and Athena

A cold shower 🥶: Why IceDB v1 was terrible

Why are you making this again?

Redesign

1. Events come in as JSON, and are written as Parquet files

2. Parquet files are merged together to reduce file count and optimize query performance

3. Parquet files are immutable, and kept track of in a consistent meta store (e.g. CockroachDB)

4. Avoid S3 ListObject calls

1. Indexing, row sorting, and file naming

2. Make it a framework that is embeddable - make a thin wrapper as an API or deeply integrate it into your stack. This allows us to make configuration easier and more flexible by writing some Python istead of arcane configuration files and SQL statements

3. Data formatting that anything can query - DuckDB, ClickHouse, Pandas, anything that can read parquet can read IceDB.

Indexing, row sorting, and file naming

Elevator pitch on parquet files

Hive partitioning

The Meta Store

Tying it together

• We keep track of higher-level information within the file name, such as some namespace, year, month, day, etc. Depending on how much data you have you may choose to omit parts like day, and have a handful of files for the month

• Inside of the file, we have columns for more granular querying. For example we might sort by event_id, then timestamp within the parquet file. This way we can more efficiently query for single events, which is so commonly done with event-based systems (or joining multiple queries of different events)

• If you know your queries are always going to be on some other dimension (like event_id), we can move that to the first Hive partition in the file name as well. This flexibility is analogous to primary keys in OLTP, or partition and ordering keys in ClickHouse

• We can choose how and when merging happens - for example we can merge all files that have a matching y, m, d in the hive partition

• When inserting, we can control how aggressively we want to batch, and write each insert as one file. This is the same as systems like ClickHouse

• We always add a unique row_id so that in the event we don’t use the meta store, we can deduplicate rows on that column

• We can dynamically change the row_group_size of the parquet files depending on how many rows exist, how many unique values are present in a column, and more

Merging Files

select partition, filename, filesize
from known_files
where active = true
and filesize < {}
order by partition {}

select filename
from known_files
where active = true
and partition = '{}'
and filename in ({})
for update

q = '''
COPY (
    select *
    from read_parquet([{}], hive_partitioning=1)
) TO 's3://testbucket/{}'
'''.format(','.join(list(map(lambda x: "'s3://testbucket/{}/{}'".format(x[0], x[1]), buf))), new_f_path)
ddb.execute(q)

update known_files
set active = false
where active = true
and partition = '{}'
and filename in ({});

insert into known_files (filename, filesize, partition)  VALUES ('{}', {}, '{}');

Custom Merge Queries

select
    user_id,
    sum(clicks) as clicks,
    gen_random_uuid()::TEXT as _row_id
from read_parquet(?, hive_partitioning=1)
group by user_id

Inserting Rows

partmap = {}
    for row in rows:
        # merge the rows into same parts
        rowTime = datetime.utcfromtimestamp(row['ts']/1000)
        part = 'y={}/m={}/d={}'.format('{}'.format(rowTime.year).zfill(4), '{}'.format(rowTime.month).zfill(2), '{}'.format(rowTime.day).zfill(2))
        if part not in partmap:
            partmap[part] = []
        partmap[part].append(row)

final_files = []
    for part in partmap:
        # upload parquet file
        filename = '{}.parquet'.format(uuid4())
        fullpath = part + '/' + filename
        final_files.append(fullpath)
        partrows = partmap[part]
        # use a DF for inserting into duckdb
        df = pa.Table.from_pydict({
            'ts': map(lambda row: row['ts'], partrows),
            'event': map(lambda row: row['event'], partrows),
            'properties': map(lambda row: json.dumps(row['properties']), partrows), # turn the properties into string
            'row_id': map(lambda row: str(uuid4()), partrows) # give every row a unique id for dedupe
        })
        ddb.sql('''
            copy (select * from df order by event, ts) to '{}'
        '''.format('s3://testbucket/' + fullpath))

# get file metadata
obj = s3.head_object(
    Bucket='testbucket',
    Key=fullpath
)
fileSize = obj['ContentLength']

# insert into meta store
rowTime = datetime.utcfromtimestamp(partrows[0]['ts'] / 1000) # this is janky
with conn.cursor() as cursor:
    cursor.execute('''
        insert into known_files (filename, filesize, partition)  VALUES ('{}', {}, '{}')
    '''.format(filename, fileSize, part))

Querying IceDB

select sum((properties::JSON->>'numtime')::int64) as agg, uid
from icedb(start_month:=2, end_month:=8)
where event = 'page_load'
group by uid
order by agg desc

1. First, it asked the meta store what files exist between 2023-02-01 and 2023-08-01, so that even if we had files for a million years, we’d only get the ones in the time range we were interested in

2. Second, DuckDB read the metadata of all known parquet files in that time range from S3, and checked if any row_groups had the page_load event in them

3. Third, DuckDB started reading the row_groups that might have the page_load event in them, stream the rows into memory just like normal query processing, reading 1 row_group per core concurrently.

4. Finally, DuckDB gives us the query result in tens of milliseconds, even if the sum of the files in that range were in TBs, it knew how to shred down the read data efficiently to only read hundreds of MB at most.

def get_files(syear: int, smonth: int, sday: int, eyear: int, emonth: int, eday: int) -> list[str]:
    with conn.cursor() as mycur:
        q = '''
        select partition, filename
        from known_files
        where active = true
        AND partition >= 'y={}/m={}/d={}'
        AND partition <= 'y={}/m={}/d={}'
        '''.format('{}'.format(syear).zfill(4), '{}'.format(smonth).zfill(2), '{}'.format(sday).zfill(2),'{}'.format(eyear).zfill(4), '{}'.format(emonth).zfill(2), '{}'.format(eday).zfill(2))
        mycur.execute(q)
        rows = mycur.fetchall()
        return list(map(lambda x: 's3://testbucket/{}/{}'.format(x[0], x[1]), rows))

ddb.create_function('get_files', get_files, [ty.INTEGER, ty.INTEGER, ty.INTEGER, ty.INTEGER, ty.INTEGER, ty.INTEGER], list[str])

ddb.sql('''
    create macro if not exists get_f(start_year:=2023, start_month:=1, start_day:=1, end_year:=2023, end_month:=1, end_day:=1) as get_files(start_year, start_month, start_day, end_year, end_month, end_day)
''')

# the `select *` gets overridden by the select on the macro, and has the same explain plan as if you did the `read_parquet` select directly. It effectively drops the unused columns.
ddb.sql('''
    create macro if not exists icedb(start_year:=2023, start_month:=1, start_day:=1, end_year:=2023, end_month:=1, end_day:=1) as table select * from read_parquet(get_files(start_year, start_month, start_day, end_year, end_month, end_day), hive_partitioning=1)
''')

Future Work

Multi-Tenancy

Conclusion