Implement support for apache/parquet-format#221 (IEEE total ordering of floating point numbers)

Author: Mytherin

extension/parquet/include/parquet_column_schema.hpp

@@ -27,6 +27,8 @@ enum class ParquetExtraTypeInfo {
 	FLOAT16
 };
 
+enum class ParquetTypeOrder { STANDARD, IEEE_754_TOTAL_ORDER };
+
 struct ParquetColumnSchema {
 	ParquetColumnSchema() = default;
 	ParquetColumnSchema(idx_t max_define, idx_t max_repeat, idx_t schema_index, idx_t file_index,
@@ -47,6 +49,7 @@ struct ParquetColumnSchema {
 	uint32_t type_scale = 0;
 	duckdb_parquet::Type::type parquet_type = duckdb_parquet::Type::INT32;
 	ParquetExtraTypeInfo type_info = ParquetExtraTypeInfo::NONE;
+	ParquetTypeOrder type_order = ParquetTypeOrder::STANDARD;
 	vector<ParquetColumnSchema> children;
 
 	unique_ptr<BaseStatistics> Stats(ParquetReader &reader, idx_t row_group_idx_p,

extension/parquet/include/parquet_reader.hpp

@@ -210,6 +210,10 @@ class ParquetReader : public BaseFileReader {
 	MultiFileColumnDefinition ParseColumnDefinition(const duckdb_parquet::FileMetaData &file_meta_data,
 	                                                ParquetColumnSchema &element);
 
+	void ParseColumnOrdersRecursive(ParquetColumnSchema &column_schema,
+	                                const duckdb_parquet::FileMetaData &file_meta_data, idx_t &column_order_idx);
+	void ParseColumnOrders(ParquetColumnSchema &column_schema, const duckdb_parquet::FileMetaData &file_meta_data);
+
 private:
 	unique_ptr<FileHandle> file_handle;
 };

extension/parquet/include/parquet_writer.hpp

@@ -150,6 +150,7 @@ class ParquetWriter {
 
 private:
 	void GatherWrittenStatistics();
+	void WriteColumnOrders(const ParquetColumnSchema &column_schemas, duckdb_parquet::FileMetaData &file_meta_data);
 
 private:
 	ClientContext &context;

extension/parquet/parquet_reader.cpp

@@ -596,6 +596,8 @@ unique_ptr<ParquetColumnSchema> ParquetReader::ParseSchema() {
 	if (root.type.id() != LogicalTypeId::STRUCT) {
 		throw InvalidInputException("Root element of Parquet file must be a struct");
 	}
+	// parse the column orders (if any)
+	ParseColumnOrders(root, *file_meta_data);
 	D_ASSERT(next_schema_idx == file_meta_data->schema.size() - 1);
 	D_ASSERT(file_meta_data->row_groups.empty() || next_file_idx == file_meta_data->row_groups[0].columns.size());
 	if (parquet_options.file_row_number) {
@@ -611,6 +613,33 @@ unique_ptr<ParquetColumnSchema> ParquetReader::ParseSchema() {
 	return make_uniq<ParquetColumnSchema>(root);
 }
 
+void ParquetReader::ParseColumnOrdersRecursive(ParquetColumnSchema &column_schema,
+                                               const duckdb_parquet::FileMetaData &file_meta_data,
+                                               idx_t &column_order_idx) {
+	if (column_order_idx >= file_meta_data.column_orders.size()) {
+		// exhausted the column order set
+		return;
+	}
+	if (column_schema.children.empty()) {
+		// root schema - read the order
+		auto &order = file_meta_data.column_orders[column_order_idx];
+		if (order.__isset.IEEE_754_TOTAL_ORDER) {
+			column_schema.type_order = ParquetTypeOrder::IEEE_754_TOTAL_ORDER;
+		}
+		column_order_idx++;
+		return;
+	}
+	for (auto &child_schema : column_schema.children) {
+		ParseColumnOrdersRecursive(child_schema, file_meta_data, column_order_idx);
+	}
+}
+
+void ParquetReader::ParseColumnOrders(ParquetColumnSchema &column_schema,
+                                      const duckdb_parquet::FileMetaData &file_meta_data) {
+	idx_t column_order_idx = 0;
+	ParseColumnOrdersRecursive(column_schema, file_meta_data, column_order_idx);
+}
+
 MultiFileColumnDefinition ParquetReader::ParseColumnDefinition(const FileMetaData &file_meta_data,
                                                                ParquetColumnSchema &element) {
 	MultiFileColumnDefinition result(element.name, element.type);

extension/parquet/parquet_statistics.cpp

@@ -54,6 +54,10 @@ unique_ptr<BaseStatistics> ParquetStatisticsUtils::CreateNumericStats(const Logi
 static unique_ptr<BaseStatistics> CreateFloatingPointStats(const LogicalType &type,
                                                            const ParquetColumnSchema &schema_ele,
                                                            const duckdb_parquet::Statistics &parquet_stats) {
+	if (schema_ele.type_order == ParquetTypeOrder::IEEE_754_TOTAL_ORDER) {
+		// if we have the total order defined we can treat floating point stats as regular numeric stats
+		return ParquetStatisticsUtils::CreateNumericStats(type, schema_ele, parquet_stats);
+	}
 	auto stats = NumericStats::CreateUnknown(type);
 
 	// floating point values can always have NaN values - hence we cannot use the max value from the file

extension/parquet/parquet_writer.cpp

@@ -417,11 +417,32 @@ ParquetWriter::ParquetWriter(ClientContext &context, FileSystem &fs, string file
 		column_writers.push_back(
 		    ColumnWriter::CreateWriterRecursive(context, *this, file_meta_data.schema, child_schema, path_in_schema));
 	}
+	for (auto &child_schema : column_schemas) {
+		WriteColumnOrders(child_schema, file_meta_data);
+	}
+	file_meta_data.__isset.column_orders = true;
 }
 
 ParquetWriter::~ParquetWriter() {
 }
 
+void ParquetWriter::WriteColumnOrders(const ParquetColumnSchema &column_schema,
+                                      duckdb_parquet::FileMetaData &file_meta_data) {
+	if (column_schema.children.empty()) {
+		// root schema - write the orders
+		duckdb_parquet::ColumnOrder order;
+		if (column_schema.type.IsFloating()) {
+			duckdb_parquet::IEEE754TotalOrder floating_order;
+			order.__set_IEEE_754_TOTAL_ORDER(floating_order);
+		}
+		file_meta_data.column_orders.push_back(order);
+		return;
+	}
+	for (auto &child_schema : column_schema.children) {
+		WriteColumnOrders(child_schema, file_meta_data);
+	}
+}
+
 void ParquetWriter::PrepareRowGroup(ColumnDataCollection &buffer, PreparedRowGroup &result) {
 	// We write 8 columns at a time so that iterating over ColumnDataCollection is more efficient
 	static constexpr idx_t COLUMNS_PER_PASS = 8;

test/sql/copy/parquet/test_parquet_fp_order.test

@@ -0,0 +1,23 @@
+# name: test/sql/copy/parquet/test_parquet_fp_order.test
+# description: Test floating point order
+# group: [parquet]
+
+require parquet
+
+statement ok
+COPY (SELECT i::DOUBLE d FROM range(10) t(i)) TO '__TEST_DIR__/fp_order.parquet';
+
+# verify that we can read back the correct stats
+query I
+SELECT stats(d) FROM '__TEST_DIR__/fp_order.parquet' LIMIT 1
+----
+<REGEX>:.*Min.*0.0.*Max.*9.0.*
+
+statement ok
+SET explain_output = PHYSICAL_ONLY
+
+# we can prune this
+query II
+EXPLAIN SELECT d FROM '__TEST_DIR__/fp_order.parquet' WHERE d>100
+----
+physical_plan	<REGEX>:.*EMPTY_RESULT.*

third_party/parquet/parquet.thrift

@@ -943,6 +943,9 @@ struct RowGroup {
 /** Empty struct to signal the order defined by the physical or logical type */
 struct TypeDefinedOrder {}
 
+/** Empty struct to signal IEEE 754 total order for floating point types */
+struct IEEE754TotalOrder {}
+
 /**
  * Union to specify the order used for the min_value and max_value fields for a
  * column. This union takes the role of an enhanced enum that allows rich
@@ -1008,6 +1011,7 @@ union ColumnOrder {
    *       `-0.0` should be written into the min statistics field.
    */
   1: TypeDefinedOrder TYPE_ORDER;
+  2: IEEE754TotalOrder IEEE_754_TOTAL_ORDER;
 }
 
 struct PageLocation {