From a867a94df4fb3b561c7ecf0225eec2d5669083bd Mon Sep 17 00:00:00 2001
From: Jan Kaul <jankaul@mailbox.org>
Date: Sat, 2 May 2026 10:17:08 -0700
Subject: [PATCH 01/10] inital implementation

---
 parquet/src/arrow/hll.rs | 35 +++++++++++++++++++++++++++++++++++
 parquet/src/arrow/mod.rs |  2 ++
 2 files changed, 37 insertions(+)
 create mode 100644 parquet/src/arrow/hll.rs

diff --git a/parquet/src/arrow/hll.rs b/parquet/src/arrow/hll.rs
new file mode 100644
index 000000000000..911b1563406e
--- /dev/null
+++ b/parquet/src/arrow/hll.rs
@@ -0,0 +1,35 @@
+use arrow_array::{
+    ArrayRef, PrimitiveArray,
+    cast::as_primitive_array,
+    types::{Int64Type, UInt16Type, UInt64Type},
+};
+use twox_hash::XxHash64;
+
+use crate::errors::ParquetError;
+
+const HLL_HASH_SEED: u64 = 0;
+
+struct HyperLogLog {
+    registers: [u8; 256],
+}
+
+impl HyperLogLog {
+    pub fn insert_array(&mut self, array: ArrayRef) -> Result<(), ParquetError> {
+        let array: &PrimitiveArray<Int64Type> = as_primitive_array(&array);
+        let hashes: PrimitiveArray<UInt64Type> =
+            array.unary(|v| XxHash64::oneshot(HLL_HASH_SEED, &v.to_le_bytes()));
+        let packed: PrimitiveArray<UInt16Type> = hashes.unary(|h| {
+            let index = (h >> 56) as u16;
+            let lz = (h << 8).leading_zeros() as u16;
+            (index << 8) | lz
+        });
+        for entry in packed.iter().flatten() {
+            let [index, lz] = entry.to_be_bytes();
+            let register = &mut self.registers[index as usize];
+            if lz > *register {
+                *register = lz;
+            }
+        }
+        Ok(())
+    }
+}
diff --git a/parquet/src/arrow/mod.rs b/parquet/src/arrow/mod.rs
index 52152988166f..7c0b542d66d9 100644
--- a/parquet/src/arrow/mod.rs
+++ b/parquet/src/arrow/mod.rs
@@ -185,6 +185,8 @@ pub mod arrow_writer;
 mod buffer;
 mod decoder;
 
+pub mod hll;
+
 #[cfg(feature = "async")]
 pub mod async_reader;
 #[cfg(feature = "async")]

From 3cd73594b02609cd204b1b9855ff7372505d0435 Mon Sep 17 00:00:00 2001
From: Jan Kaul <jankaul@mailbox.org>
Date: Sat, 2 May 2026 10:27:54 -0700
Subject: [PATCH 02/10] use parallel computation

---
 parquet/Cargo.toml       |  4 +++-
 parquet/src/arrow/hll.rs | 27 +++++++++++++++------------
 2 files changed, 18 insertions(+), 13 deletions(-)

diff --git a/parquet/Cargo.toml b/parquet/Cargo.toml
index d1ada01c3773..3c5ffce0a315 100644
--- a/parquet/Cargo.toml
+++ b/parquet/Cargo.toml
@@ -37,10 +37,12 @@ ring = { version = "0.17", default-features = false, features = ["wasm32_unknown
 ahash = { version = "0.8", default-features = false, features = ["runtime-rng"] }
 
 [dependencies]
+arrow-arith = { workspace = true, optional = true }
 arrow-array = { workspace = true, optional = true }
 arrow-buffer = { workspace = true, optional = true }
 arrow-csv = { workspace = true, optional = true }
 arrow-data = { workspace = true, optional = true }
+arrow-ord = { workspace = true, optional = true }
 arrow-schema = { workspace = true, optional = true }
 arrow-select = { workspace = true, optional = true }
 arrow-ipc = { workspace = true, optional = true }
@@ -104,7 +106,7 @@ default = ["arrow", "snap", "brotli", "flate2-zlib-rs", "lz4", "zstd", "base64",
 # Enable lz4
 lz4 = ["lz4_flex"]
 # Enable arrow reader/writer APIs
-arrow = ["base64", "arrow-array", "arrow-buffer", "arrow-data", "arrow-schema", "arrow-select", "arrow-ipc"]
+arrow = ["base64", "arrow-arith", "arrow-array", "arrow-buffer", "arrow-data", "arrow-ord", "arrow-schema", "arrow-select", "arrow-ipc"]
 # Enable support for arrow canonical extension types
 arrow_canonical_extension_types = ["arrow-schema?/canonical_extension_types"]
 # Enable CLI tools
diff --git a/parquet/src/arrow/hll.rs b/parquet/src/arrow/hll.rs
index 911b1563406e..dfed53818a21 100644
--- a/parquet/src/arrow/hll.rs
+++ b/parquet/src/arrow/hll.rs
@@ -1,7 +1,7 @@
 use arrow_array::{
-    ArrayRef, PrimitiveArray,
+    ArrayRef, PrimitiveArray, UInt8Array,
     cast::as_primitive_array,
-    types::{Int64Type, UInt16Type, UInt64Type},
+    types::{Int64Type, UInt8Type, UInt64Type},
 };
 use twox_hash::XxHash64;
 
@@ -18,16 +18,19 @@ impl HyperLogLog {
         let array: &PrimitiveArray<Int64Type> = as_primitive_array(&array);
         let hashes: PrimitiveArray<UInt64Type> =
             array.unary(|v| XxHash64::oneshot(HLL_HASH_SEED, &v.to_le_bytes()));
-        let packed: PrimitiveArray<UInt16Type> = hashes.unary(|h| {
-            let index = (h >> 56) as u16;
-            let lz = (h << 8).leading_zeros() as u16;
-            (index << 8) | lz
-        });
-        for entry in packed.iter().flatten() {
-            let [index, lz] = entry.to_be_bytes();
-            let register = &mut self.registers[index as usize];
-            if lz > *register {
-                *register = lz;
+        let indices: UInt8Array = hashes.unary(|h| (h >> 56) as u8);
+        let leading_zeros: UInt8Array = hashes.unary(|h| (h << 8).leading_zeros() as u8);
+
+        for i in 0..=255u8 {
+            let scalar = UInt8Array::new_scalar(i);
+            let mask = arrow_ord::cmp::eq(&indices, &scalar)?;
+            let filtered = arrow_select::filter::filter(&leading_zeros, &mask)?;
+            let filtered: &PrimitiveArray<UInt8Type> = as_primitive_array(&filtered);
+            if let Some(max_lz) = arrow_arith::aggregate::max(filtered) {
+                let register = &mut self.registers[i as usize];
+                if max_lz > *register {
+                    *register = max_lz;
+                }
             }
         }
         Ok(())

From 3eac7b7875308b785db6d13c4f9f11e51e1361e8 Mon Sep 17 00:00:00 2001
From: Jan Kaul <jankaul@mailbox.org>
Date: Sat, 2 May 2026 10:32:57 -0700
Subject: [PATCH 03/10] count method

---
 parquet/src/arrow/hll.rs | 47 +++++++++++++++++++++++++++++++++++++++-
 parquet/src/arrow/mod.rs |  1 +
 2 files changed, 47 insertions(+), 1 deletion(-)

diff --git a/parquet/src/arrow/hll.rs b/parquet/src/arrow/hll.rs
index dfed53818a21..ebead42a751f 100644
--- a/parquet/src/arrow/hll.rs
+++ b/parquet/src/arrow/hll.rs
@@ -1,3 +1,7 @@
+//! Approximate distinct-value counting via the [HyperLogLog] algorithm.
+//!
+//! [HyperLogLog]: https://en.wikipedia.org/wiki/HyperLogLog
+
 use arrow_array::{
     ArrayRef, PrimitiveArray, UInt8Array,
     cast::as_primitive_array,
@@ -9,11 +13,23 @@ use crate::errors::ParquetError;
 
 const HLL_HASH_SEED: u64 = 0;
 
-struct HyperLogLog {
+/// HyperLogLog sketch with `m = 256` registers (precision `p = 8`).
+///
+/// Each `Int64` value inserted is hashed with xxHash64; the top 8 bits select
+/// the register and the remaining 56 bits contribute their leading-zero count.
+/// Each register tracks the maximum leading-zero count observed for its bucket,
+/// and [`count`](Self::count) derives a cardinality estimate from those values.
+pub struct HyperLogLog {
     registers: [u8; 256],
 }
 
 impl HyperLogLog {
+    /// Insert every non-null value of `array` (must be `Int64`) into the sketch.
+    ///
+    /// Each value is hashed once; for every register `i` the new maximum
+    /// leading-zero count among hashes routed to bucket `i` is merged into
+    /// `registers[i]`. Returns an error if `array` is not an `Int64Array` or if
+    /// the underlying Arrow compute kernels fail.
     pub fn insert_array(&mut self, array: ArrayRef) -> Result<(), ParquetError> {
         let array: &PrimitiveArray<Int64Type> = as_primitive_array(&array);
         let hashes: PrimitiveArray<UInt64Type> =
@@ -35,4 +51,33 @@ impl HyperLogLog {
         }
         Ok(())
     }
+
+    /// Estimate the number of distinct values inserted so far.
+    ///
+    /// Uses the standard HyperLogLog estimator
+    /// `E = α_m · m² / Σ 2^(-M[j])` (harmonic-mean form) with the
+    /// bias-correction constant `α_m` for `m = 256`. When the raw estimate is
+    /// small (`≤ 2.5 m`) and at least one register is still zero, falls back to
+    /// linear counting `m · ln(m / V)`, which is more accurate at low
+    /// cardinalities. See Flajolet et al., *HyperLogLog: the analysis of a
+    /// near-optimal cardinality estimation algorithm* (2007).
+    pub fn count(&self) -> f64 {
+        const M: usize = 256;
+        const M_F: f64 = M as f64;
+        const ALPHA: f64 = 0.7213 / (1.0 + 1.079 / M_F);
+
+        let z: f64 = self
+            .registers
+            .iter()
+            .map(|&r| (-(r as f64)).exp2())
+            .sum();
+        let raw = ALPHA * M_F * M_F / z;
+
+        let zeros = self.registers.iter().filter(|&&r| r == 0).count();
+        if raw <= 2.5 * M_F && zeros > 0 {
+            M_F * (M_F / zeros as f64).ln()
+        } else {
+            raw
+        }
+    }
 }
diff --git a/parquet/src/arrow/mod.rs b/parquet/src/arrow/mod.rs
index 7c0b542d66d9..85860d9d07c4 100644
--- a/parquet/src/arrow/mod.rs
+++ b/parquet/src/arrow/mod.rs
@@ -187,6 +187,7 @@ mod decoder;
 
 pub mod hll;
 
+
 #[cfg(feature = "async")]
 pub mod async_reader;
 #[cfg(feature = "async")]

From e50aa7b24c005610c6acbee1176e88e262086011 Mon Sep 17 00:00:00 2001
From: Jan Kaul <jankaul@mailbox.org>
Date: Sat, 2 May 2026 10:36:38 -0700
Subject: [PATCH 04/10] fmt

---
 parquet/src/arrow/hll.rs | 6 +-----
 parquet/src/arrow/mod.rs | 1 -
 2 files changed, 1 insertion(+), 6 deletions(-)

diff --git a/parquet/src/arrow/hll.rs b/parquet/src/arrow/hll.rs
index ebead42a751f..f330deda8b4c 100644
--- a/parquet/src/arrow/hll.rs
+++ b/parquet/src/arrow/hll.rs
@@ -66,11 +66,7 @@ impl HyperLogLog {
         const M_F: f64 = M as f64;
         const ALPHA: f64 = 0.7213 / (1.0 + 1.079 / M_F);
 
-        let z: f64 = self
-            .registers
-            .iter()
-            .map(|&r| (-(r as f64)).exp2())
-            .sum();
+        let z: f64 = self.registers.iter().map(|&r| (-(r as f64)).exp2()).sum();
         let raw = ALPHA * M_F * M_F / z;
 
         let zeros = self.registers.iter().filter(|&&r| r == 0).count();
diff --git a/parquet/src/arrow/mod.rs b/parquet/src/arrow/mod.rs
index 85860d9d07c4..7c0b542d66d9 100644
--- a/parquet/src/arrow/mod.rs
+++ b/parquet/src/arrow/mod.rs
@@ -187,7 +187,6 @@ mod decoder;
 
 pub mod hll;
 
-
 #[cfg(feature = "async")]
 pub mod async_reader;
 #[cfg(feature = "async")]

From 04ea091bb5c2b87efdfee7cd6afb9fe2697c4f37 Mon Sep 17 00:00:00 2001
From: Jan Kaul <jankaul@mailbox.org>
Date: Sat, 2 May 2026 13:41:07 -0700
Subject: [PATCH 05/10] write distinct-count metadata

---
 parquet/src/arrow/arrow_writer/mod.rs | 100 ++++++++++++++++++++++++--
 parquet/src/arrow/hll.rs              |  36 +++++++---
 parquet/src/column/writer/mod.rs      |  14 ++++
 parquet/src/file/properties.rs        |  31 ++++++++
 4 files changed, 164 insertions(+), 17 deletions(-)

diff --git a/parquet/src/arrow/arrow_writer/mod.rs b/parquet/src/arrow/arrow_writer/mod.rs
index 979988eebc05..b3533ce14731 100644
--- a/parquet/src/arrow/arrow_writer/mod.rs
+++ b/parquet/src/arrow/arrow_writer/mod.rs
@@ -35,6 +35,8 @@ use super::schema::{add_encoded_arrow_schema_to_metadata, decimal_length_from_pr
 
 use crate::arrow::ArrowSchemaConverter;
 use crate::arrow::arrow_writer::byte_array::ByteArrayEncoder;
+use crate::arrow::hll::HyperLogLog;
+use crate::basic::Type as PhysicalType;
 use crate::column::page::{CompressedPage, PageWriteSpec, PageWriter};
 use crate::column::page_encryption::PageEncryptor;
 use crate::column::writer::encoder::ColumnValueEncoder;
@@ -863,14 +865,17 @@ impl std::fmt::Debug for ArrowColumnWriter {
 
 enum ArrowColumnWriterImpl {
     ByteArray(GenericColumnWriter<'static, ByteArrayEncoder>),
-    Column(ColumnWriter<'static>),
+    Column(ColumnWriter<'static>, Option<HyperLogLog>),
 }
 
 impl ArrowColumnWriter {
     /// Write an [`ArrowLeafColumn`]
     pub fn write(&mut self, col: &ArrowLeafColumn) -> Result<()> {
         match &mut self.writer {
-            ArrowColumnWriterImpl::Column(c) => {
+            ArrowColumnWriterImpl::Column(c, hll) => {
+                if let Some(hll) = hll.as_mut() {
+                    hll.insert_array(col.0.array().clone())?;
+                }
                 let leaf = col.0.array();
                 match leaf.as_any_dictionary_opt() {
                     Some(dictionary) => {
@@ -892,7 +897,14 @@ impl ArrowColumnWriter {
     pub fn close(self) -> Result<ArrowColumnChunk> {
         let close = match self.writer {
             ArrowColumnWriterImpl::ByteArray(c) => c.close()?,
-            ArrowColumnWriterImpl::Column(c) => c.close()?,
+            ArrowColumnWriterImpl::Column(mut c, hll) => {
+                if let Some(hll) = hll {
+                    if let ColumnWriter::Int64ColumnWriter(int64) = &mut c {
+                        int64.set_column_distinct_count(Some(hll.count() as u64));
+                    }
+                }
+                c.close()?
+            }
         };
         let chunk = Arc::try_unwrap(self.chunk).ok().unwrap();
         let data = chunk.into_inner().unwrap();
@@ -912,7 +924,7 @@ impl ArrowColumnWriter {
     pub fn memory_size(&self) -> usize {
         match &self.writer {
             ArrowColumnWriterImpl::ByteArray(c) => c.memory_size(),
-            ArrowColumnWriterImpl::Column(c) => c.memory_size(),
+            ArrowColumnWriterImpl::Column(c, _) => c.memory_size(),
         }
     }
 
@@ -926,7 +938,7 @@ impl ArrowColumnWriter {
     pub fn get_estimated_total_bytes(&self) -> usize {
         match &self.writer {
             ArrowColumnWriterImpl::ByteArray(c) => c.get_estimated_total_bytes() as _,
-            ArrowColumnWriterImpl::Column(c) => c.get_estimated_total_bytes() as _,
+            ArrowColumnWriterImpl::Column(c, _) => c.get_estimated_total_bytes() as _,
         }
     }
 }
@@ -1134,9 +1146,12 @@ impl ArrowColumnWriterFactory {
             let page_writer = self.create_page_writer(desc, out.len())?;
             let chunk = page_writer.buffer.clone();
             let writer = get_column_writer(desc.clone(), props.clone(), page_writer);
+            let hll = (props.estimate_distinct_count()
+                && desc.physical_type() == PhysicalType::INT64)
+                .then(HyperLogLog::new);
             Ok(ArrowColumnWriter {
                 chunk,
-                writer: ArrowColumnWriterImpl::Column(writer),
+                writer: ArrowColumnWriterImpl::Column(writer, hll),
             })
         };
 
@@ -4955,4 +4970,77 @@ mod tests {
         let total_rows: i64 = sizes.iter().sum();
         assert_eq!(total_rows, 100, "Total rows should be preserved");
     }
+
+    fn distinct_count_from_buffer(buffer: Vec<u8>, column: usize) -> Option<u64> {
+        let reader = SerializedFileReader::new(Bytes::from(buffer)).unwrap();
+        reader
+            .metadata()
+            .row_group(0)
+            .column(column)
+            .statistics()
+            .and_then(|s| s.distinct_count_opt())
+    }
+
+    fn write_to_buffer(batch: &RecordBatch, props: WriterProperties) -> Vec<u8> {
+        let mut buffer = Vec::new();
+        let mut writer = ArrowWriter::try_new(&mut buffer, batch.schema(), Some(props)).unwrap();
+        writer.write(batch).unwrap();
+        writer.close().unwrap();
+        buffer
+    }
+
+    #[test]
+    fn estimate_distinct_count_int64() {
+        let n_distinct: i64 = 10_000;
+        let schema = Arc::new(Schema::new(vec![Field::new("a", DataType::Int64, false)]));
+        let values: Int64Array = (0..n_distinct).collect();
+        let batch = RecordBatch::try_new(schema, vec![Arc::new(values)]).unwrap();
+
+        let props = WriterProperties::builder()
+            .set_estimate_distinct_count(true)
+            .build();
+        let buffer = write_to_buffer(&batch, props);
+
+        let count = distinct_count_from_buffer(buffer, 0).expect("distinct_count populated");
+        let n = n_distinct as u64;
+        let lower = n - n / 10;
+        let upper = n + n / 10;
+        assert!(
+            count >= lower && count <= upper,
+            "HLL estimate {count} not within ±10% of {n}",
+        );
+    }
+
+    #[test]
+    fn estimate_distinct_count_disabled_by_default() {
+        let schema = Arc::new(Schema::new(vec![Field::new("a", DataType::Int64, false)]));
+        let values: Int64Array = (0..1_000_i64).collect();
+        let batch = RecordBatch::try_new(schema, vec![Arc::new(values)]).unwrap();
+
+        let buffer = write_to_buffer(&batch, WriterProperties::builder().build());
+        assert_eq!(distinct_count_from_buffer(buffer, 0), None);
+    }
+
+    #[test]
+    fn estimate_distinct_count_only_int64() {
+        let schema = Arc::new(Schema::new(vec![
+            Field::new("i32", DataType::Int32, false),
+            Field::new("i64", DataType::Int64, false),
+        ]));
+        let i32_values = Int32Array::from((0..500).collect::<Vec<_>>());
+        let i64_values: Int64Array = (0..500_i64).collect();
+        let batch = RecordBatch::try_new(
+            schema,
+            vec![Arc::new(i32_values), Arc::new(i64_values)],
+        )
+        .unwrap();
+
+        let props = WriterProperties::builder()
+            .set_estimate_distinct_count(true)
+            .build();
+        let buffer = write_to_buffer(&batch, props);
+
+        assert_eq!(distinct_count_from_buffer(buffer.clone(), 0), None);
+        assert!(distinct_count_from_buffer(buffer, 1).is_some());
+    }
 }
diff --git a/parquet/src/arrow/hll.rs b/parquet/src/arrow/hll.rs
index f330deda8b4c..dda743c793f1 100644
--- a/parquet/src/arrow/hll.rs
+++ b/parquet/src/arrow/hll.rs
@@ -16,36 +16,50 @@ const HLL_HASH_SEED: u64 = 0;
 /// HyperLogLog sketch with `m = 256` registers (precision `p = 8`).
 ///
 /// Each `Int64` value inserted is hashed with xxHash64; the top 8 bits select
-/// the register and the remaining 56 bits contribute their leading-zero count.
-/// Each register tracks the maximum leading-zero count observed for its bucket,
-/// and [`count`](Self::count) derives a cardinality estimate from those values.
+/// the register and the remaining 56 bits contribute their *rank* — the
+/// 1-indexed position of the leftmost 1-bit, i.e. `leading_zeros + 1`. Each
+/// register tracks the maximum rank observed for its bucket, and
+/// [`count`](Self::count) derives a cardinality estimate from those values.
 pub struct HyperLogLog {
     registers: [u8; 256],
 }
 
+impl Default for HyperLogLog {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
 impl HyperLogLog {
+    /// Create an empty sketch with all registers set to zero.
+    pub fn new() -> Self {
+        Self {
+            registers: [0; 256],
+        }
+    }
+
     /// Insert every non-null value of `array` (must be `Int64`) into the sketch.
     ///
     /// Each value is hashed once; for every register `i` the new maximum
-    /// leading-zero count among hashes routed to bucket `i` is merged into
-    /// `registers[i]`. Returns an error if `array` is not an `Int64Array` or if
-    /// the underlying Arrow compute kernels fail.
+    /// rank among hashes routed to bucket `i` is merged into `registers[i]`.
+    /// Returns an error if `array` is not an `Int64Array` or if the underlying
+    /// Arrow compute kernels fail.
     pub fn insert_array(&mut self, array: ArrayRef) -> Result<(), ParquetError> {
         let array: &PrimitiveArray<Int64Type> = as_primitive_array(&array);
         let hashes: PrimitiveArray<UInt64Type> =
             array.unary(|v| XxHash64::oneshot(HLL_HASH_SEED, &v.to_le_bytes()));
         let indices: UInt8Array = hashes.unary(|h| (h >> 56) as u8);
-        let leading_zeros: UInt8Array = hashes.unary(|h| (h << 8).leading_zeros() as u8);
+        let ranks: UInt8Array = hashes.unary(|h| (h << 8).leading_zeros() as u8 + 1);
 
         for i in 0..=255u8 {
             let scalar = UInt8Array::new_scalar(i);
             let mask = arrow_ord::cmp::eq(&indices, &scalar)?;
-            let filtered = arrow_select::filter::filter(&leading_zeros, &mask)?;
+            let filtered = arrow_select::filter::filter(&ranks, &mask)?;
             let filtered: &PrimitiveArray<UInt8Type> = as_primitive_array(&filtered);
-            if let Some(max_lz) = arrow_arith::aggregate::max(filtered) {
+            if let Some(max_rank) = arrow_arith::aggregate::max(filtered) {
                 let register = &mut self.registers[i as usize];
-                if max_lz > *register {
-                    *register = max_lz;
+                if max_rank > *register {
+                    *register = max_rank;
                 }
             }
         }
diff --git a/parquet/src/column/writer/mod.rs b/parquet/src/column/writer/mod.rs
index c014397f132e..f2bc8527a792 100644
--- a/parquet/src/column/writer/mod.rs
+++ b/parquet/src/column/writer/mod.rs
@@ -424,6 +424,20 @@ impl<'a, E: ColumnValueEncoder> GenericColumnWriter<'a, E> {
         }
     }
 
+    /// Override the `distinct_count` value that will be written into this
+    /// column chunk's statistics.
+    ///
+    /// Normally [`Self::write_batch_with_statistics`] only honors a
+    /// `distinct_count` argument on the very first batch written to a chunk,
+    /// and clears it on subsequent writes. This setter writes the given value
+    /// directly into the chunk's accumulated metrics so callers that compute
+    /// the count externally (for example via a HyperLogLog sketch fed by every
+    /// batch) can install the final value at any point before
+    /// [`Self::close`].
+    pub fn set_column_distinct_count(&mut self, count: Option<u64>) {
+        self.column_metrics.column_distinct_count = count;
+    }
+
     #[allow(clippy::too_many_arguments)]
     pub(crate) fn write_batch_internal(
         &mut self,
diff --git a/parquet/src/file/properties.rs b/parquet/src/file/properties.rs
index ae21de304404..a00eb4a0eb59 100644
--- a/parquet/src/file/properties.rs
+++ b/parquet/src/file/properties.rs
@@ -61,6 +61,8 @@ pub const DEFAULT_STATISTICS_TRUNCATE_LENGTH: Option<usize> = Some(64);
 pub const DEFAULT_OFFSET_INDEX_DISABLED: bool = false;
 /// Default values for [`WriterProperties::coerce_types`]
 pub const DEFAULT_COERCE_TYPES: bool = false;
+/// Default value for [`WriterProperties::estimate_distinct_count`]
+pub const DEFAULT_ESTIMATE_DISTINCT_COUNT: bool = false;
 
 /// Parquet writer version.
 ///
@@ -168,6 +170,7 @@ pub struct WriterProperties {
     column_index_truncate_length: Option<usize>,
     statistics_truncate_length: Option<usize>,
     coerce_types: bool,
+    estimate_distinct_count: bool,
     #[cfg(feature = "encryption")]
     pub(crate) file_encryption_properties: Option<Arc<FileEncryptionProperties>>,
 }
@@ -364,6 +367,15 @@ impl WriterProperties {
         self.coerce_types
     }
 
+    /// Returns `true` if the writer should populate the `distinct_count`
+    /// statistic for `Int64` columns using a HyperLogLog estimate.
+    ///
+    /// For more details see
+    /// [`WriterPropertiesBuilder::set_estimate_distinct_count`].
+    pub fn estimate_distinct_count(&self) -> bool {
+        self.estimate_distinct_count
+    }
+
     /// Returns encoding for a data page, when dictionary encoding is enabled.
     ///
     /// This is not configurable.
@@ -487,6 +499,7 @@ pub struct WriterPropertiesBuilder {
     column_index_truncate_length: Option<usize>,
     statistics_truncate_length: Option<usize>,
     coerce_types: bool,
+    estimate_distinct_count: bool,
     #[cfg(feature = "encryption")]
     file_encryption_properties: Option<Arc<FileEncryptionProperties>>,
 }
@@ -510,6 +523,7 @@ impl Default for WriterPropertiesBuilder {
             column_index_truncate_length: DEFAULT_COLUMN_INDEX_TRUNCATE_LENGTH,
             statistics_truncate_length: DEFAULT_STATISTICS_TRUNCATE_LENGTH,
             coerce_types: DEFAULT_COERCE_TYPES,
+            estimate_distinct_count: DEFAULT_ESTIMATE_DISTINCT_COUNT,
             #[cfg(feature = "encryption")]
             file_encryption_properties: None,
         }
@@ -535,6 +549,7 @@ impl WriterPropertiesBuilder {
             column_index_truncate_length: self.column_index_truncate_length,
             statistics_truncate_length: self.statistics_truncate_length,
             coerce_types: self.coerce_types,
+            estimate_distinct_count: self.estimate_distinct_count,
             #[cfg(feature = "encryption")]
             file_encryption_properties: self.file_encryption_properties,
         }
@@ -750,6 +765,21 @@ impl WriterPropertiesBuilder {
         self
     }
 
+    /// Enable HyperLogLog-based `distinct_count` estimation for `Int64`
+    /// columns (defaults to `false` via [`DEFAULT_ESTIMATE_DISTINCT_COUNT`]).
+    ///
+    /// When enabled, the arrow writer feeds every `Int64` array routed to a
+    /// column chunk into a HyperLogLog sketch (m=256, p=8) and stores
+    /// `count() as u64` as the chunk's `distinct_count` statistic. The estimate
+    /// has a standard error of roughly 6.5%; consumers that require an exact
+    /// count must not rely on it.
+    ///
+    /// Has no effect on non-`Int64` columns.
+    pub fn set_estimate_distinct_count(mut self, value: bool) -> Self {
+        self.estimate_distinct_count = value;
+        self
+    }
+
     /// Sets FileEncryptionProperties (defaults to `None`)
     #[cfg(feature = "encryption")]
     pub fn with_file_encryption_properties(
@@ -1033,6 +1063,7 @@ impl From<WriterProperties> for WriterPropertiesBuilder {
             column_index_truncate_length: props.column_index_truncate_length,
             statistics_truncate_length: props.statistics_truncate_length,
             coerce_types: props.coerce_types,
+            estimate_distinct_count: props.estimate_distinct_count,
             #[cfg(feature = "encryption")]
             file_encryption_properties: props.file_encryption_properties,
         }

From 0a9773c84dd070ad02ef06b4adb577955e1e7726 Mon Sep 17 00:00:00 2001
From: Jan Kaul <jankaul@mailbox.org>
Date: Sat, 2 May 2026 14:05:36 -0700
Subject: [PATCH 06/10] rename property

---
 parquet/src/arrow/arrow_writer/mod.rs | 12 ++++++------
 parquet/src/file/properties.rs        | 26 +++++++++++++-------------
 2 files changed, 19 insertions(+), 19 deletions(-)

diff --git a/parquet/src/arrow/arrow_writer/mod.rs b/parquet/src/arrow/arrow_writer/mod.rs
index b3533ce14731..dbc8652eb274 100644
--- a/parquet/src/arrow/arrow_writer/mod.rs
+++ b/parquet/src/arrow/arrow_writer/mod.rs
@@ -1146,7 +1146,7 @@ impl ArrowColumnWriterFactory {
             let page_writer = self.create_page_writer(desc, out.len())?;
             let chunk = page_writer.buffer.clone();
             let writer = get_column_writer(desc.clone(), props.clone(), page_writer);
-            let hll = (props.estimate_distinct_count()
+            let hll = (props.estimate_int64_distinct_count()
                 && desc.physical_type() == PhysicalType::INT64)
                 .then(HyperLogLog::new);
             Ok(ArrowColumnWriter {
@@ -4990,14 +4990,14 @@ mod tests {
     }
 
     #[test]
-    fn estimate_distinct_count_int64() {
+    fn estimate_int64_distinct_count_int64() {
         let n_distinct: i64 = 10_000;
         let schema = Arc::new(Schema::new(vec![Field::new("a", DataType::Int64, false)]));
         let values: Int64Array = (0..n_distinct).collect();
         let batch = RecordBatch::try_new(schema, vec![Arc::new(values)]).unwrap();
 
         let props = WriterProperties::builder()
-            .set_estimate_distinct_count(true)
+            .set_estimate_int64_distinct_count(true)
             .build();
         let buffer = write_to_buffer(&batch, props);
 
@@ -5012,7 +5012,7 @@ mod tests {
     }
 
     #[test]
-    fn estimate_distinct_count_disabled_by_default() {
+    fn estimate_int64_distinct_count_disabled_by_default() {
         let schema = Arc::new(Schema::new(vec![Field::new("a", DataType::Int64, false)]));
         let values: Int64Array = (0..1_000_i64).collect();
         let batch = RecordBatch::try_new(schema, vec![Arc::new(values)]).unwrap();
@@ -5022,7 +5022,7 @@ mod tests {
     }
 
     #[test]
-    fn estimate_distinct_count_only_int64() {
+    fn estimate_int64_distinct_count_only_int64() {
         let schema = Arc::new(Schema::new(vec![
             Field::new("i32", DataType::Int32, false),
             Field::new("i64", DataType::Int64, false),
@@ -5036,7 +5036,7 @@ mod tests {
         .unwrap();
 
         let props = WriterProperties::builder()
-            .set_estimate_distinct_count(true)
+            .set_estimate_int64_distinct_count(true)
             .build();
         let buffer = write_to_buffer(&batch, props);
 
diff --git a/parquet/src/file/properties.rs b/parquet/src/file/properties.rs
index a00eb4a0eb59..c8bceb33eaf1 100644
--- a/parquet/src/file/properties.rs
+++ b/parquet/src/file/properties.rs
@@ -61,8 +61,8 @@ pub const DEFAULT_STATISTICS_TRUNCATE_LENGTH: Option<usize> = Some(64);
 pub const DEFAULT_OFFSET_INDEX_DISABLED: bool = false;
 /// Default values for [`WriterProperties::coerce_types`]
 pub const DEFAULT_COERCE_TYPES: bool = false;
-/// Default value for [`WriterProperties::estimate_distinct_count`]
-pub const DEFAULT_ESTIMATE_DISTINCT_COUNT: bool = false;
+/// Default value for [`WriterProperties::estimate_int64_distinct_count`]
+pub const DEFAULT_ESTIMATE_INT64_DISTINCT_COUNT: bool = false;
 
 /// Parquet writer version.
 ///
@@ -170,7 +170,7 @@ pub struct WriterProperties {
     column_index_truncate_length: Option<usize>,
     statistics_truncate_length: Option<usize>,
     coerce_types: bool,
-    estimate_distinct_count: bool,
+    estimate_int64_distinct_count: bool,
     #[cfg(feature = "encryption")]
     pub(crate) file_encryption_properties: Option<Arc<FileEncryptionProperties>>,
 }
@@ -371,9 +371,9 @@ impl WriterProperties {
     /// statistic for `Int64` columns using a HyperLogLog estimate.
     ///
     /// For more details see
-    /// [`WriterPropertiesBuilder::set_estimate_distinct_count`].
-    pub fn estimate_distinct_count(&self) -> bool {
-        self.estimate_distinct_count
+    /// [`WriterPropertiesBuilder::set_estimate_int64_distinct_count`].
+    pub fn estimate_int64_distinct_count(&self) -> bool {
+        self.estimate_int64_distinct_count
     }
 
     /// Returns encoding for a data page, when dictionary encoding is enabled.
@@ -499,7 +499,7 @@ pub struct WriterPropertiesBuilder {
     column_index_truncate_length: Option<usize>,
     statistics_truncate_length: Option<usize>,
     coerce_types: bool,
-    estimate_distinct_count: bool,
+    estimate_int64_distinct_count: bool,
     #[cfg(feature = "encryption")]
     file_encryption_properties: Option<Arc<FileEncryptionProperties>>,
 }
@@ -523,7 +523,7 @@ impl Default for WriterPropertiesBuilder {
             column_index_truncate_length: DEFAULT_COLUMN_INDEX_TRUNCATE_LENGTH,
             statistics_truncate_length: DEFAULT_STATISTICS_TRUNCATE_LENGTH,
             coerce_types: DEFAULT_COERCE_TYPES,
-            estimate_distinct_count: DEFAULT_ESTIMATE_DISTINCT_COUNT,
+            estimate_int64_distinct_count: DEFAULT_ESTIMATE_INT64_DISTINCT_COUNT,
             #[cfg(feature = "encryption")]
             file_encryption_properties: None,
         }
@@ -549,7 +549,7 @@ impl WriterPropertiesBuilder {
             column_index_truncate_length: self.column_index_truncate_length,
             statistics_truncate_length: self.statistics_truncate_length,
             coerce_types: self.coerce_types,
-            estimate_distinct_count: self.estimate_distinct_count,
+            estimate_int64_distinct_count: self.estimate_int64_distinct_count,
             #[cfg(feature = "encryption")]
             file_encryption_properties: self.file_encryption_properties,
         }
@@ -766,7 +766,7 @@ impl WriterPropertiesBuilder {
     }
 
     /// Enable HyperLogLog-based `distinct_count` estimation for `Int64`
-    /// columns (defaults to `false` via [`DEFAULT_ESTIMATE_DISTINCT_COUNT`]).
+    /// columns (defaults to `false` via [`DEFAULT_ESTIMATE_INT64_DISTINCT_COUNT`]).
     ///
     /// When enabled, the arrow writer feeds every `Int64` array routed to a
     /// column chunk into a HyperLogLog sketch (m=256, p=8) and stores
@@ -775,8 +775,8 @@ impl WriterPropertiesBuilder {
     /// count must not rely on it.
     ///
     /// Has no effect on non-`Int64` columns.
-    pub fn set_estimate_distinct_count(mut self, value: bool) -> Self {
-        self.estimate_distinct_count = value;
+    pub fn set_estimate_int64_distinct_count(mut self, value: bool) -> Self {
+        self.estimate_int64_distinct_count = value;
         self
     }
 
@@ -1063,7 +1063,7 @@ impl From<WriterProperties> for WriterPropertiesBuilder {
             column_index_truncate_length: props.column_index_truncate_length,
             statistics_truncate_length: props.statistics_truncate_length,
             coerce_types: props.coerce_types,
-            estimate_distinct_count: props.estimate_distinct_count,
+            estimate_int64_distinct_count: props.estimate_int64_distinct_count,
             #[cfg(feature = "encryption")]
             file_encryption_properties: props.file_encryption_properties,
         }

From 1fe281f7bac7550e201fac70ea26227b9584417b Mon Sep 17 00:00:00 2001
From: Jan Kaul <jankaul@mailbox.org>
Date: Sat, 2 May 2026 14:07:32 -0700
Subject: [PATCH 07/10] fmt

---
 parquet/src/arrow/arrow_writer/mod.rs | 7 ++-----
 1 file changed, 2 insertions(+), 5 deletions(-)

diff --git a/parquet/src/arrow/arrow_writer/mod.rs b/parquet/src/arrow/arrow_writer/mod.rs
index dbc8652eb274..8c92b8f4ef4c 100644
--- a/parquet/src/arrow/arrow_writer/mod.rs
+++ b/parquet/src/arrow/arrow_writer/mod.rs
@@ -5029,11 +5029,8 @@ mod tests {
         ]));
         let i32_values = Int32Array::from((0..500).collect::<Vec<_>>());
         let i64_values: Int64Array = (0..500_i64).collect();
-        let batch = RecordBatch::try_new(
-            schema,
-            vec![Arc::new(i32_values), Arc::new(i64_values)],
-        )
-        .unwrap();
+        let batch =
+            RecordBatch::try_new(schema, vec![Arc::new(i32_values), Arc::new(i64_values)]).unwrap();
 
         let props = WriterProperties::builder()
             .set_estimate_int64_distinct_count(true)

From 9723d9e728b86d1dfe6e9d560929c2274575eb5a Mon Sep 17 00:00:00 2001
From: Jan Kaul <jankaul@mailbox.org>
Date: Sat, 2 May 2026 14:08:44 -0700
Subject: [PATCH 08/10] fix header

---
 parquet/src/arrow/hll.rs | 17 +++++++++++++++++
 1 file changed, 17 insertions(+)

diff --git a/parquet/src/arrow/hll.rs b/parquet/src/arrow/hll.rs
index dda743c793f1..44326c543cb9 100644
--- a/parquet/src/arrow/hll.rs
+++ b/parquet/src/arrow/hll.rs
@@ -1,3 +1,20 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
 //! Approximate distinct-value counting via the [HyperLogLog] algorithm.
 //!
 //! [HyperLogLog]: https://en.wikipedia.org/wiki/HyperLogLog

From dfd2140ab53bf17865c3dca8a2e39139bb8240d0 Mon Sep 17 00:00:00 2001
From: Jan Kaul <jankaul@mailbox.org>
Date: Sat, 2 May 2026 16:33:48 -0700
Subject: [PATCH 09/10] fix ci

---
 parquet/Cargo.toml | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/parquet/Cargo.toml b/parquet/Cargo.toml
index 3c5ffce0a315..269734e5a539 100644
--- a/parquet/Cargo.toml
+++ b/parquet/Cargo.toml
@@ -51,7 +51,7 @@ parquet-variant = { workspace = true, optional = true }
 parquet-variant-json = { workspace = true, optional = true }
 parquet-variant-compute = { workspace = true, optional = true }
 
-object_store = { version = "0.13.1", default-features = false, optional = true }
+object_store = { version = "0.13.1", default-features = false, optional = true, features = ["tokio"] }
 
 bytes = { version = "1.1", default-features = false, features = ["std"] }
 thrift = { version = "0.17", default-features = false }

From 1e02d305643c57998b99b1f2b9d41e3272a04d5e Mon Sep 17 00:00:00 2001
From: Jan Kaul <jankaul@mailbox.org>
Date: Sat, 2 May 2026 18:04:16 -0700
Subject: [PATCH 10/10] use key value metadata

---
 parquet/src/arrow/arrow_writer/mod.rs | 45 +++++++++++++++++++++------
 parquet/src/file/properties.rs        | 31 ------------------
 2 files changed, 36 insertions(+), 40 deletions(-)

diff --git a/parquet/src/arrow/arrow_writer/mod.rs b/parquet/src/arrow/arrow_writer/mod.rs
index 8c92b8f4ef4c..048d64272bbe 100644
--- a/parquet/src/arrow/arrow_writer/mod.rs
+++ b/parquet/src/arrow/arrow_writer/mod.rs
@@ -37,6 +37,30 @@ use crate::arrow::ArrowSchemaConverter;
 use crate::arrow::arrow_writer::byte_array::ByteArrayEncoder;
 use crate::arrow::hll::HyperLogLog;
 use crate::basic::Type as PhysicalType;
+
+/// File-level KV metadata key that opts in to HyperLogLog-based `distinct_count`
+/// estimation for `Int64` columns. When `WriterProperties::key_value_metadata()`
+/// contains an entry with this key whose value is `"true"`, the arrow writer
+/// feeds every `Int64` array routed to a column chunk into a HyperLogLog sketch
+/// (m=256, p=8) and stores the estimate in the chunk's `distinct_count`
+/// statistic. The estimate has a standard error of roughly 6.5%; consumers that
+/// require an exact count must not rely on it.
+///
+/// Has no effect on non-`Int64` columns.
+const ICEBERG_ESTIMATE_INT64_DISTINCT_COUNT_META_KEY: &str =
+    "iceberg.estimate-int64-distinct-count";
+
+fn estimate_int64_distinct_count_enabled(props: &WriterProperties) -> bool {
+    props
+        .key_value_metadata()
+        .map(|kv| {
+            kv.iter().any(|entry| {
+                entry.key == ICEBERG_ESTIMATE_INT64_DISTINCT_COUNT_META_KEY
+                    && entry.value.as_deref() == Some("true")
+            })
+        })
+        .unwrap_or(false)
+}
 use crate::column::page::{CompressedPage, PageWriteSpec, PageWriter};
 use crate::column::page_encryption::PageEncryptor;
 use crate::column::writer::encoder::ColumnValueEncoder;
@@ -1146,7 +1170,7 @@ impl ArrowColumnWriterFactory {
             let page_writer = self.create_page_writer(desc, out.len())?;
             let chunk = page_writer.buffer.clone();
             let writer = get_column_writer(desc.clone(), props.clone(), page_writer);
-            let hll = (props.estimate_int64_distinct_count()
+            let hll = (estimate_int64_distinct_count_enabled(props)
                 && desc.physical_type() == PhysicalType::INT64)
                 .then(HyperLogLog::new);
             Ok(ArrowColumnWriter {
@@ -4989,6 +5013,15 @@ mod tests {
         buffer
     }
 
+    fn estimate_int64_distinct_count_props() -> WriterProperties {
+        WriterProperties::builder()
+            .set_key_value_metadata(Some(vec![KeyValue::new(
+                ICEBERG_ESTIMATE_INT64_DISTINCT_COUNT_META_KEY.to_owned(),
+                "true".to_owned(),
+            )]))
+            .build()
+    }
+
     #[test]
     fn estimate_int64_distinct_count_int64() {
         let n_distinct: i64 = 10_000;
@@ -4996,10 +5029,7 @@ mod tests {
         let values: Int64Array = (0..n_distinct).collect();
         let batch = RecordBatch::try_new(schema, vec![Arc::new(values)]).unwrap();
 
-        let props = WriterProperties::builder()
-            .set_estimate_int64_distinct_count(true)
-            .build();
-        let buffer = write_to_buffer(&batch, props);
+        let buffer = write_to_buffer(&batch, estimate_int64_distinct_count_props());
 
         let count = distinct_count_from_buffer(buffer, 0).expect("distinct_count populated");
         let n = n_distinct as u64;
@@ -5032,10 +5062,7 @@ mod tests {
         let batch =
             RecordBatch::try_new(schema, vec![Arc::new(i32_values), Arc::new(i64_values)]).unwrap();
 
-        let props = WriterProperties::builder()
-            .set_estimate_int64_distinct_count(true)
-            .build();
-        let buffer = write_to_buffer(&batch, props);
+        let buffer = write_to_buffer(&batch, estimate_int64_distinct_count_props());
 
         assert_eq!(distinct_count_from_buffer(buffer.clone(), 0), None);
         assert!(distinct_count_from_buffer(buffer, 1).is_some());
diff --git a/parquet/src/file/properties.rs b/parquet/src/file/properties.rs
index c8bceb33eaf1..ae21de304404 100644
--- a/parquet/src/file/properties.rs
+++ b/parquet/src/file/properties.rs
@@ -61,8 +61,6 @@ pub const DEFAULT_STATISTICS_TRUNCATE_LENGTH: Option<usize> = Some(64);
 pub const DEFAULT_OFFSET_INDEX_DISABLED: bool = false;
 /// Default values for [`WriterProperties::coerce_types`]
 pub const DEFAULT_COERCE_TYPES: bool = false;
-/// Default value for [`WriterProperties::estimate_int64_distinct_count`]
-pub const DEFAULT_ESTIMATE_INT64_DISTINCT_COUNT: bool = false;
 
 /// Parquet writer version.
 ///
@@ -170,7 +168,6 @@ pub struct WriterProperties {
     column_index_truncate_length: Option<usize>,
     statistics_truncate_length: Option<usize>,
     coerce_types: bool,
-    estimate_int64_distinct_count: bool,
     #[cfg(feature = "encryption")]
     pub(crate) file_encryption_properties: Option<Arc<FileEncryptionProperties>>,
 }
@@ -367,15 +364,6 @@ impl WriterProperties {
         self.coerce_types
     }
 
-    /// Returns `true` if the writer should populate the `distinct_count`
-    /// statistic for `Int64` columns using a HyperLogLog estimate.
-    ///
-    /// For more details see
-    /// [`WriterPropertiesBuilder::set_estimate_int64_distinct_count`].
-    pub fn estimate_int64_distinct_count(&self) -> bool {
-        self.estimate_int64_distinct_count
-    }
-
     /// Returns encoding for a data page, when dictionary encoding is enabled.
     ///
     /// This is not configurable.
@@ -499,7 +487,6 @@ pub struct WriterPropertiesBuilder {
     column_index_truncate_length: Option<usize>,
     statistics_truncate_length: Option<usize>,
     coerce_types: bool,
-    estimate_int64_distinct_count: bool,
     #[cfg(feature = "encryption")]
     file_encryption_properties: Option<Arc<FileEncryptionProperties>>,
 }
@@ -523,7 +510,6 @@ impl Default for WriterPropertiesBuilder {
             column_index_truncate_length: DEFAULT_COLUMN_INDEX_TRUNCATE_LENGTH,
             statistics_truncate_length: DEFAULT_STATISTICS_TRUNCATE_LENGTH,
             coerce_types: DEFAULT_COERCE_TYPES,
-            estimate_int64_distinct_count: DEFAULT_ESTIMATE_INT64_DISTINCT_COUNT,
             #[cfg(feature = "encryption")]
             file_encryption_properties: None,
         }
@@ -549,7 +535,6 @@ impl WriterPropertiesBuilder {
             column_index_truncate_length: self.column_index_truncate_length,
             statistics_truncate_length: self.statistics_truncate_length,
             coerce_types: self.coerce_types,
-            estimate_int64_distinct_count: self.estimate_int64_distinct_count,
             #[cfg(feature = "encryption")]
             file_encryption_properties: self.file_encryption_properties,
         }
@@ -765,21 +750,6 @@ impl WriterPropertiesBuilder {
         self
     }
 
-    /// Enable HyperLogLog-based `distinct_count` estimation for `Int64`
-    /// columns (defaults to `false` via [`DEFAULT_ESTIMATE_INT64_DISTINCT_COUNT`]).
-    ///
-    /// When enabled, the arrow writer feeds every `Int64` array routed to a
-    /// column chunk into a HyperLogLog sketch (m=256, p=8) and stores
-    /// `count() as u64` as the chunk's `distinct_count` statistic. The estimate
-    /// has a standard error of roughly 6.5%; consumers that require an exact
-    /// count must not rely on it.
-    ///
-    /// Has no effect on non-`Int64` columns.
-    pub fn set_estimate_int64_distinct_count(mut self, value: bool) -> Self {
-        self.estimate_int64_distinct_count = value;
-        self
-    }
-
     /// Sets FileEncryptionProperties (defaults to `None`)
     #[cfg(feature = "encryption")]
     pub fn with_file_encryption_properties(
@@ -1063,7 +1033,6 @@ impl From<WriterProperties> for WriterPropertiesBuilder {
             column_index_truncate_length: props.column_index_truncate_length,
             statistics_truncate_length: props.statistics_truncate_length,
             coerce_types: props.coerce_types,
-            estimate_int64_distinct_count: props.estimate_int64_distinct_count,
             #[cfg(feature = "encryption")]
             file_encryption_properties: props.file_encryption_properties,
         }