[Format] Passing column statistics through Arrow C data interface

[ianmcook]

Describe the enhancement requested

Is there any standard or convention for passing column statistics through the C data interface?

For example, say there is a module that reads a Parquet file into memory in Arrow format then passes the data arrays to another module through the C data interface. If the Parquet file metadata includes Parquet column statistics such as distinct_count, max, min, and null_count, can the sending module pass those statistics through the C data interface, to allow the receiving module to use the statistics to perform computations more efficiently?

Component(s)

Format

[zeroshade]

There's no current standard convention, there's a few ways such could be sent though such as via a struct array with one row and fields for each of those or various other configurations.

Passing such statistics would be beyond the scope of the current C Data interface IMHO

[ianmcook]

cc @Tishj @pdet

There is a related discussion at duckdb/duckdb#4636 (reply in thread)

[pdet]

There's no current standard convention, there's a few ways such could be sent though such as via a struct array with one row and fields for each of those or various other configurations.

Passing such statistics would be beyond the scope of the current C Data interface IMHO

I would argue that it makes sense to include statistical information in the C data interface. Efficient execution of complex queries requires statistical information, and I believe that most Arrow Producers possess this information. Therefore, it should be somehow passed. An alternative I can think of for the C-Data interface is to wrap the statistical information in top-level objects (e.g., ArrowRecordBatch in Python), but that approach is quite cumbersome and would necessitate specific implementations for every client API.

[pdet]

Just to provide a clearer example of how statistics can be useful for query optimization.

In DuckDB, join ordering is currently determined using heuristics based on table cardinalities with future plans to enhance this with sample statistics.

Not only join ordering is affected by statistics but even the choice of the probe side in a hash join, will be determined based on the expected cardinalities.

One example of a query that is affected by join ordering is Q 21 of tpch. However, the plan for it is too big to share it easily in a GitHub Discussion.

To give a simpler example of how cardinalities affect this, I've created two tables.

1. t - Has 10^8 rows, ranging from 0 to 10^8
2. t_2 - Has 10 rows, ranging from 0 to 10

My example query is a simple inner join of these two tables and we calculate the sum of t.i.

SELECT SUM(t.i) from t inner join t_2 on (t_2.k = t.i)

Because the optimizer doesn't have any information of statistics from the Arrow side, it will basically pick the probe side depending on what's presented in the query.

SELECT SUM(t.i) from t_2 inner join t on (t_2.k = t.i)

This would result in a slightly different plan, yet with significant differences in performance.

As depicted in the screenshot of executing both queries, choosing the incorrect probe side for this query already results in a performance difference of an order of magnitude. For more complex queries, the variations in execution time could be not only larger but also more difficult to trace.

For reference, the code I used for this example:

import duckdb
import time
import statistics

con = duckdb.connect()

# Create table with 10^8
con.execute("CREATE TABLE t as SELECT * FROM RANGE(0, 100000000) tbl(i)")
# Create Table with 10
con.execute("CREATE TABLE t_2 as SELECT * FROM RANGE(0, 10) tbl(k)")

query_slow = '''
SELECT SUM(t.i) from t inner join t_2 on (t_2.k = t.i);
'''

query_fast = '''
SELECT SUM(t.i) from t_2 inner join t on (t_2.k = t.i);'''

t = con.execute("FROM t").fetch_arrow_table()
t_2 = con.execute("FROM t_2").fetch_arrow_table()

con_2 = duckdb.connect()

print("DuckDB Arrow - Query Slow")

print(con_2.execute("EXPLAIN " + query_slow).fetchall()[0][1])

execution_times = []

for _ in range(5):
    start_time = time.time()
    con_2.execute(query_slow)
    end_time = time.time()
    
    execution_times.append(end_time - start_time)

median_time = statistics.median(execution_times)

print(median_time)

print("DuckDB Arrow - Query Fast")

print(con_2.execute("EXPLAIN " + query_fast).fetchall()[0][1])

execution_times = []

for _ in range(5):
    start_time = time.time()
    con_2.execute(query_fast)
    end_time = time.time()
    
    execution_times.append(end_time - start_time)

median_time = statistics.median(execution_times)

print(median_time)

[kou]

I'm considering some approaches for this use case. This is not completed yet but share my idea so far. Feedback is appreciated.

ADBC uses the following schema to return statistics:

https://github.com/apache/arrow-adbc/blob/3f80831d12b6e5a78a4321f67e28d652951241cf/adbc.h#L1739-L1778

It's designed for returning statistics of a database.

We can simplify this schema because we can just return statistics of a record batch. For example:

Field Name	Field Type	Comments
column_name	utf8	(1)
statistic_key	int16 not null	(2)
statistic_value	VALUE_SCHEMA not null
statistic_is_approximate	bool not null	(3)

1. If null, then the statistic applies to the entire table.
2. A dictionary-encoded statistic name (although we do not use the Arrow
   dictionary type). Values in [0, 1024) are reserved for ADBC. Other
   values are for implementation-specific statistics. For the definitions
   of predefined statistic types, see adbc-table-statistics. To get
   driver-specific statistic names, use AdbcConnectionGetStatisticNames().
3. If true, then the value is approximate or best-effort.

VALUE_SCHEMA is a dense union with members:

Field Name	Field Type
int64	int64
uint64	uint64
float64	float64
binary	binary

TODO: How to represent statistic key? Should we use ADBC style? (Assigning an ID for each statistic key and using it.)

If we represent statistics as a record batch, we can pass statistics through Arrow C data interface. This may be a reasonable approach.

If we use this approach, we need to do the followings:

• Define a schema as a specification
• Add statistics related APIs to Apache Arrow C++ and other implementation because we need two more implementations for specification change
  • https://arrow.apache.org/docs/format/Changing.html#at-least-two-reference-implementations
  • This is not a format change but it's better that we should follow the rule
  • We can work on this for Apache Arrow C++ before we propose a specification because statistics will be useful for general propose
• Apache Arrow C++: Add support for importing statistics from Apache Parquet C++
• ...

TODO: Consider statistics related API for Apache Arrow C++.

[zeroshade]

I'd be curious what others think of this approach as opposed to actually making a format change to include statistics alongside the record batches in the API. Particular in the case of a stream of batches.

I'm not against it, I just don't know if others would be opposed to needing an entirely separate record batch being sent containing the statistics

[ianmcook]

I'd be curious what others think of this approach as opposed to actually making a format change to include statistics alongside the record batches in the API

I think the top priority should be to avoid breaking ABI compatibility. I suspect that most users of the C data interface will not want to pass statistics. We should avoid doing anything that would cause disruption for those users.

[kou]

Ah, I should have written an approach that changes the current Arrow C data interface.
I'll write it later.

[mapleFU]

If the Parquet file metadata includes Parquet column statistics such as distinct_count, max, min, and null_count, can the sending module pass those statistics through the C data interface, to allow the receiving module to use the statistics to perform computations more efficiently?

This proposal is great. Just a un-related issue, Parquet DistinctCount might merely used currently. And now Dataset Scanner might parsing the min-max / null as Expression

[mapleFU]

#38837 (comment)

Is {u}int64/float64/string the only supported types? What would the min-max schema being when facing the different types?

[lidavidm]

At least for ADBC, the idea was that other types can be encoded in those choices. (Decimals can be represented as bytes, dates can be represented as int64, etc.)

[kou]

Some approaches that are based the C Data interface https://arrow.apache.org/docs/format/CDataInterface.html :

(1) Add get_statistics callback to ArrowArray

For example:

struct ArrowArray {
  // Array data description
  int64_t length;
  int64_t null_count;
  int64_t offset;
  int64_t n_buffers;
  int64_t n_children;
  const void** buffers;
  struct ArrowArray** children;
  struct ArrowArray* dictionary;

  // Callback to return statistics of this ArrowArray
  struct ArrowArray *(*get_statistics)(struct ArrowArray*);
  // Release callback
  void (*release)(struct ArrowArray*);
  // Opaque producer-specific data
  void* private_data;
};

This uses a struct ArrowArray to represent statistics like #38837 (comment) but we can define struct ArrowStatistics or something instead.

Note that this is a backward incompatible change. struct ArrowArray doesn't have version information nor spaces for extension. We can't do this without breaking backward compatibility.

(2) Add statistics to ArrowSchema::metadata

https://arrow.apache.org/docs/format/CDataInterface.html#c.ArrowSchema.metadata

If we choose this approach, we will preserve some metadata key such as ARROW:XXX like we did for IPC format: https://arrow.apache.org/docs/format/Columnar.html#custom-application-metadata

Here are some ideas how to put statistics into ArrowSchema::metadata:

1. Use struct ArrowArray* (pointer) as ARROW:statistics metadata value
2. Use multiple metadata to represent statistics

Here is an example for the 2. approach:

{
  "ARROW:statistics:column1:max": 2.9,
  "ARROW:statistics:column1:max:approximate": true,
  "ARROW:statistics:column2:average_byte_width": 29.9
}

TODO:

• How to encode each value (2.9, true and 29.9) to raw byte data? We can use only raw byte data for a value of ArrowSchema::metadata.
• Can we support same name columns with this approach?
• This isn't space effective because we have many duplicated texts such as ARROW:statistics:.

Note that this is a (roughly) backward compatible change. I think that most users don't use ARROW:XXX as metadata key.

This may not work with the C stream interface https://arrow.apache.org/docs/format/CStreamInterface.html . Because it shares one struct ArrowSchema with multiple struct ArrowArray. Each struct ArrowArray will have different statistics.

[lidavidm]

Do consumers want per-batch metadata anyways? I would assume in the context of say DuckDB is that they'd like to get statistics for the whole stream up front, without reading any data, and use that to inform their query plan.