Column Statistics

[ChongWei905]

I've read the ISSUE about Column Statistics: #4163 and the recent merged PR about zonemap: #4244
It feels like the zonemap codes could also solve fragment level column statistics generating. So I wonder if there's a plan on doing that? And if there is column statistics, where would it be written in?
I'm currently working on using lance file reader and writer to add a new format for iceberg. I'm not sure if it's acceptable to add the column statistics to the footer of a lance file just like parquet file?
Or maybe could i generate the column statistics during the process of lance file writing. And expose a method to get these statistics before the writer is closed?

[HaochengLIU]

• It feels like the zonemap codes could also solve fragment level column statistics generating. So I wonder if there's a plan on doing that? And if there is column statistics, where would it be written in?

Yep, the stats will be there once you build the zonemap index. The data is written into an auxiliary index file at table layer like all other scalar indices.

Say I have below test case.
Input:
* Fragment 1: - 10 rows   -> 0  -> 9
* Fragment 2: - 7 rows    -> 10 -> 16
* Fragment 3: - 4 rows    -> 20 -> 23
* Zone size AKA “rows_per_zone” (from user) - 5

Output:
fragment id | min | max | zone_size
1           | 0   |  4  | 5
1           | 5   |  9  | 5
2           | 10  | 14  | 5
2           | 15  | 16  | 2
3           | 20  | 23  | 4
Note: For simplicity, i omit null/nan count here

• I'm currently working on using lance file reader and writer to add a new format for iceberg. I'm not sure if it's acceptable to add the column statistics to the footer of a lance file just like parquet file?
  Or maybe could i generate the column statistics during the process of lance file writing. And expose a method to get these statistics before the writer is closed?

The stats used to exist at lance file level, but it was removed recently due to requirement changes. Could you share more details on your usecases so folks can chime in?
cc @westonpace @jackye1995

[westonpace]

It feels like the zonemap codes could also solve fragment level column statistics generating. So I wonder if there's a plan on doing that?

Yes. We can use zone map indexes as an initial source of column statistics. Partly I think we might want something else someday that is even smaller and lighter than zone map (e.g. just one set of min/max/some unique values/etc. for the entire column) which we can compute for every column. Or maybe we just compute zone maps for every column or maybe we don't worry about statistics on every column since they're only really needed on columns we are filtering with and we probably want indexes on those filter columns anyways 🤷

The important part will probably just be getting an API up for column statistics and we can use zone maps to develop that API and refactor later as we learn.

I'm currently working on using lance file reader and writer to add a new format for iceberg. I'm not sure if it's acceptable to add the column statistics to the footer of a lance file just like parquet file?

Sure. You can put pretty much whatever arbitrary buffers you want into a lance file (see https://github.com/lancedb/lance/blob/60711f360b7f8692df44a0e84c98c8fdff2897a3/rust/lance-file/src/v2/writer.rs#L491). However, if you want it to be visible by the lance table format, we would probably need to do some more work. The table format does have a pretty generic "index store" concept which might be able to map to "the indexes are in file footers" or maybe we need some changes. Also, the zone map index currently expects the entire index to be in a single file. There are other indexes which support "parts" like the btree index or maybe a "partitioned zone map" is just another index type.

Just to explain some of the lance motivation here, we are trying to avoid having a column index spread into many files because then it becomes a cold start problem when you need to open potentially thousands of files (if you have billions of rows) to read what is tens of MB of data at most.

Or maybe could i generate the column statistics during the process of lance file writing. And expose a method to get these statistics before the writer is closed?

Could this be done in the iceberg writer? You can look at each batch before you send it to lance and update statistics and then, before calling the finish method, you can add them as a global buffer. Or do you think there is some change you would need in the lance file writer?

If you're interested in reusing the Rust code to collect statistics then I think we can do that too. We can make some of the utilities in the zone map index more general purpose "zone map calculating tools". IIRC we are using some datafusion internals to do the actual comparisons and aggregation.

Slightly related / tangential: There's a discussion here around how we plan to handle statistics on write in Lance (though we are using separate index file).

[ChongWei905]

Thanks a lot! I'm not sure if there's an api in java to get statistics info from arrow vector just like apache-datafusion in rust.
So I'm very interested in reusing the Rust code to collect statistics.
And if it's a common requirement to get statistics during writing. If it is, can I add an api to update statistics for the writer? And expose an api to get statistics before writer is closed.
Something like this commit (not finished yet, only update currently): main...ChongWei905:lance:main

[ChongWei905]

And by the way, is there some way to get the column size when the writer is closed?
if i accumulate the size of arrow array every writing batch, does it actually equals to the real column size in the lance file?

[westonpace]

And if it's a common requirement to get statistics during writing. If it is, can I add an api to update statistics for the writer? And expose an api to get statistics before writer is closed.

I think an API like this is probably fine. However, we can maybe make it more generic middleware, not necessarily tied to statistics...

pub trait WriterMiddleware {
  // Can also have this method return Result<()> to avoid needing to clone `RecordBatch`
  // I think that will work for stats, which just needs to observe.  Not sure if we will ever want
  // "transformational" middleware
  async fn write_batch(&mut self, batch: &RecordBatch) -> Result<RecordBatch>;
  async fn on_finish(&mut self, writer: &mut FileWriter) -> Result<()>;
}

[westonpace]

And by the way, is there some way to get the column size when the writer is closed?

Not from the writer but the reader has a method for this: https://github.com/lancedb/lance/blob/7480c9cb2d12977f41521346cfe5add971fb015e/rust/lance-file/src/v2/reader.rs#L387

if i accumulate the size of arrow array every writing batch, does it actually equals to the real column size in the lance file?

No. We perform compression (starting in 2.1) and so the size of the arrow array should generally be larger than the size of the column in the lance file. Columns are written in pages. In the above reader method we go through and sum up the size of all pages for a column to get the column size.

[ChongWei905]

I think an API like this is probably fine. However, we can maybe make it more generic middleware, not necessarily tied to statistics...

Thanks a lot, I'll try this way and create a PR soon.

[westonpace]

I've been thinking about column statistics recently and wanted to sketch up my current thinking around a possible design. It doesn't replace plan_splits (above) but could be used in addition to it.

Motivation (why indexes aren't enough)

• Column statistics should be safe enough to enable by default (even for bloom filter we might not want to do this as an index)
• Column statistics can provide more coarse-grained resolution
• Not all indexes make sense as column statistics (e.g. btree, bitmap)
• Column statistics are always present for compacted files (indexes must be updated separately)
• Statistics can be available for query planning while we assume indexes are evaluated during query execution (this is for DF compute negine. Different engines are free to handle this differently. For example, in spark / distributed query engines the startup time of a query is generally large enough that something like plan_splits might be fine to run during "planning" of the query)

Implementation

I think from above my key goal would be that column statistics are on-by-default, coarse grained, and always present once compaction has run. Given this a design could be...

• Every table has a single table statistics file, with statistics for all columns. This should probably just be a lance file. It needs to be column-projectable (so we only have to load/cache column statistics if the column is used by the query). The rows could just be fragments or we could investigate something like very large zones if people want to create jumbo fragments. However, the zone size should be large enough that the statistics are small enough to cache in memory for the entire dataset (on a per-column basis). Given this I think a good starting point would be 1M rows per zone or so.
• When a fragment is first written, statistics could be calculated and could stored in the footer of the fragment's data files (as arbitrary global buffers). If we are using a lance file to store statistics then we could just serialize the lance file into a buffer and put it in the footer of the data file that way. I'm not really sure how useful this is. I think table-level statistics, calculated at compaction time, would be considerably more useful. Engines relying on these per-file statistics are likely to be inefficient and have to deal with opening many files.
• During compaction, the statistics should be calculated for the compacted files. This means we might "no-op compact" a file even if it is already large enough, because it does not have statistics. In this case we'd read the file, and extract the statistics, but not re-write the file. The per-file footer statistics could be useful here. Otherwise, assuming normal compaction, we will just collect statistics as we write the compacted file and the per-file footer statistics of the small files are not used / needed.
• When a compaction is committed we update the table statistics. This means that concurrent compaction operations now conflict. I don't see this as a big problem because we already kind of have this constraint due to the frag reuse index and also because it's rare to run compaction operations concurrently (distributed compaction is ok, as long as it is part of one commit).

[wjones127]

I've been dreaming of a way to keep indexes up-to-date, which seems somewhat related to this discussion if we just want to use zone maps for statistics. Seems worth sharing. Here's what I've been thinking:

• Imagine you have a dataset with an IVF_PQ index and a ZoneMap index. It has three fragments, and one index segment for each index.

• Let's say you append two new fragments, potentially in different transactions. To keep the indices up-to-date, we create new index segments within each fragment as global buffers. These index segments should be referenced directly from the index section of the manifest.
  • By only creating new segments for new fragments, and not trying to update the existing indexes, you avoid any write conflicts.

• Then we run optimize indices (or really, "index segment compaction" 😄) which merges the index segments

[wjones127]

However, the manifest writing overhead may be less negligible. Manifests can already get fairly large as we get to thousands of fragments. I agree zone maps should be fairly small but if there are many columns and it is on by default then I'm not sure the manifest writing overhead (which I think has an impact on transactions-per-second) may not be negligible.

+1 on keeping manifests small whenever possible. The way I think about it, I'd like to keep the manifest < 64 KB, so we can get it in just one cloud request.

It's also helpful for writes, since we often need to retry writing the manifest when committing, and that might be much slower if we make the manifest too large.

[fangbo]

Can I share my thoughts ?

If the primary purpose of Column Statistics is to assist the query engine in optimizing execution plans, then we can follow the following principles when implementing column statistics:

• Column Statistics must be at the Dataset level.
• Column Statistics should minimize adding extra complexity and performance overhead to current writes and commits.
• Not all Columns need statistics; we need to control which fields require statistical information.
• Statistics does not need to be completely accurate, but there should be tools to refresh the statistics.
• The ability to compute statistics in a distributed manner is required. That is, statistics can be computed at the Fragment granularity, and then merged into Dataset-level statistics.

Based on the above principles, I have implemented a prototype. The following is the core design:

• A configuration lance.stats.enabled is saved in the Field's Metadata to control whether the field collects statistics.
• Statistics are saved separately. A subdirectory named _stats is created under the Dataset's base directory, and statistics are stored in this directory.
• Fragment-level Column Statistics are saved in the file _stats/colstats_frag_{frag_id}_{field_id}.lance
• Dataset-level Column Statistics are saved in the file _stats/colstats_ds_{version_id}_{field_id}.lance
• The data structure for each field's statistics is:

pub struct ColumnStatistics {
    pub field_id: u32,
    pub min: Option<ScalarValue>,
    pub max: Option<ScalarValue>,
    pub null_count: Option<u64>,
    pub nan_count: Option<u64>,
    pub distinct_count: Option<u64>,
    pub avg_len: Option<u64>,
    pub max_len: Option<u64>,
    pub hll_registers: Option<Vec<u8>>, // serialized HLL registers
}

• FileFragment provides a method named analyze_column_statistics to collect field statistics for that Fragment.

pub trait FragmentStatsExt {
    async fn analyze_column_statistics(&self, fields: Vec<Field>) -> Result<Vec<ColumnStatistics>>;
}

• Dataset provides a method named analyze_column_statistics to collect Column Statistics for the Dataset. This method calls FileFragment.analyze_column_statistics to compute Fragment-level statistics, then merges them to generate Dataset-level statistics.
• Dataset provides a method named column_statistics to return a list of Dataset-level field statistics. This method reads the contents of the file _stats/colstats_ds_{version_id}_{field_id}.lance.

pub trait DatasetStatsExt {
    async fn analyze_column_statistics(&self) -> Result<Vec<ColumnStatistics>>;

    async fn column_statistics(&self) -> Result<Vec<ColumnStatistics>>;
}

• When writing a new Fragment, statistics are not actively refreshed. Users can proactively call Dataset.analyze_column_statistics to refresh statistics as needed. The purpose of this is to ensure that statistics do not introduce additional complexity and performance overhead to current data writes.

This is the prototype PR: #5432

[westonpace]

Let's collect the requirements.

• Easily available at plan time
• Always up-to-date
• Avoid introducing new concepts
• Avoid increasing size of manifest
• Avoid introducing write conflicts on append
• Don't require per-framgent reads to gather statistics for a column
• Statistics for individual columns must be readable
• Statistics should be flexible
• Statistics should be small enough to default to on for all columns

My original solution fails "always up-to-date" and "avoid introducing new concepts". I think Jack's proposal may struggle on "avoid increasing size of manifest"

I like Will's input. I think combining Jack and Will's proposals could work. If we want "avoid introducing write conflicts on append" then we can't have a single file. However, we can interpret "don't require per-framgent reads to gather statistics for a column" to allow "one read plus one read per uncompacted fragment" then we can tolerate index fragments.

So...

• On write of a new fragment, the statistics are collected and written into an index fragment (this can be in the file footer or in a separate file)
• This index is a single-file / multi-index file. In other words, these are just indexes, but we allow multiple indexes to share the same file.
• Any index can opt-in to "always up-to-date" mode where we calculate the values on write.
• Any index can opt-in to "accessible at scan time" where we deem the index is small enough to read (and cache) at scan time. It's unclear to me if the scan time will read just the compacted index or will look into uncompacted index fragments too (this can be a flag, we can play with it).

Some challenges that will need to be overcome...

• Need to introduce some concept of multi-index file
• If all columns get some basic coarse-grained indexes turned on by default then the "multiple indexes per column" case will be much more common (we needed to handle this anyways)
• Scanner will need to figure out some rules for what indexes can be loaded at plan time and setup a cache

[zhangyue19921010]

I like Will's input. I think combining Jack and Will's proposals could work. If we want "avoid introducing write conflicts on append" then we can't have a single file. However, we can interpret "don't require per-framgent reads to gather statistics for a column" to allow "one read plus one read per uncompacted fragment" then we can tolerate index fragments.

Maybe this is more of a Merge-on-Read (MoR) way that has a base file plus multiple delta files, with the base and delta files merged during reading.

That said, I have a question（maybe a fool question 😆）: Does this design imply that statistics need to be computable and mergeable at the metadata level—without relying on rescanning data files? For example, During query, you’d take the minimum value across multiple min statistics. However, for certain metrics like average (avg), you’d instead need to store the sum and count (since you can’t derive the final average directly from multiple individual averages).

[westonpace]

Based on the above principles, I have implemented a prototype. The following is the core design:

• A configuration lance.stats.enabled is saved in the Field's Metadata to control whether the field collects statistics.
• Statistics are saved separately. A subdirectory named _stats is created under the Dataset's base directory, and statistics are stored in this directory.
• Fragment-level Column Statistics are saved in the file stats/colstats_frag{frag_id}_{field_id}.lance
• Dataset-level Column Statistics are saved in the file stats/colstats_ds{version_id}_{field_id}.lance

I'm not sure that we can have a single dataset statistics file _stats/colstats_ds_{version_id}_{field_id}.lance and handle concurrent writes without causing those writes having conflicts. My original plan was to just let this file be out of date and calculated at compaction time. Jack's plan was to put the newer stats in the manifest and move them out at compaction time. Will's plan was to give up on a single file and let uncompacted fragment have their own index fragments.

How do you propose that file _stats/colstats_ds_{version_id}_{field_id}.lance stays up to date?

[fangbo]

@westonpace

Thank you for your reply. I still have a few questions I would like to discuss:

1. Do data changes (DELETE/INSERT) commits and column-statistics updates have to occur in the same transaction ?
   In other words, must column statistics always be perfectly consistent with the data changes ?
   If the sole purpose of column statistics is to provide information to the optimizer, I would argue that they do not need to be updated in the same transaction as the data change, for the following reasons:

   • Computing column statistics can be expensive and would increase commit latency.
   • It couples two functional modules too tightly, raising overall system complexity.
   • We can leave the decision of when to refresh column statistics to the user. A typical pattern is: an analytical job finishes, the user explicitly launches a separate task to refresh statistics.

2. Should we extend ZoneMapIndex to support column statistics?
   My view is that ZoneMapIndex and column statistics are two different concepts:

   • First, ZoneMapIndex is defined per fragment and partitions the data into zones of a user-defined row count. Its two key properties are: (a) fragment-level scope, (b) variable zone size. Column statistics do not require either property.
   • Second, column statistics should be dataset-level (similar to table-level stats in other database systems). Internally we may compute them per fragment and then merge, but the public API should return dataset-wide statistics.
   • Third, the use-cases differ. ZoneMapIndex is used at query time for data pruning, whereas column statistics are used by the optimizer to choose better execution plans. Users may well create both a ZoneMapIndex and column statistics on the same column; we therefore need a way to distinguish the two kinds of index data.

[fangbo]

How do you propose that file stats/colstats_ds{version_id}_{field_id}.lance stays up to date?

My initial thought (may be narrow and foolish) is that data changes and column-statistics collection are two independent operations.
In the typical use-case, after the data has been modified the user explicitly runs a statistics-collection job that computes the new stats from the latest data and writes them to the metadata files of the corresponding dataset version.

[HaochengLIU]

(updated)
Hey there, here are my thoughts.

• If the sole purpose of column statistics is to provide information to the optimizer,

In my vision, it would also be used for workload distribution and execution time for various engines. My preference is it should be up to date, but I'm on the fence of the must requirement.

• Should we extend ZoneMapIndex to support column statistics?

When weston and I designed the ZoneMap, we deliberately make the underlying building blocks extensible so that it could be used for other similar features. No extension is needed here. Column statistics do not require the two properties, but they are the potentially important building block candidates.
Yes I share the same view that the public API should return dataset-wide/table level statistics and that's the plan. No one has disputed that in the discussion. Jack's proposal - plan_splits and with_split - is what i have in mind.

• Third, the use-cases differ. ZoneMapIndex is used at query time for data pruning, whereas column statistics are used by the optimizer to choose better execution plans. Users may well create both a ZoneMapIndex and column statistics on the same column; we therefore need a way to distinguish the two kinds of index data.

I'm confused by this part. It's by nature two different concepts and presented to users in totally separated manners.