extra log

Author: dantengsky

Cargo.lock

@@ -12,6 +12,7 @@ databend-common-expression = { workspace = true }
 databend-storages-common-table-meta = { workspace = true }
 parking_lot = { workspace = true }
 parquet = { workspace = true }
+log = "0.4.27"
 
 [build-dependencies]
 

src/query/storages/common/blocks/Cargo.toml

@@ -35,6 +35,7 @@ use parquet::file::properties::WriterProperties;
 use parquet::file::properties::WriterVersion;
 use parquet::format::FileMetaData;
 use parquet::schema::types::ColumnPath;
+use log::info;
 
 /// Disable dictionary encoding once the NDV-to-row ratio is greater than this threshold.
 const HIGH_CARDINALITY_RATIO_THRESHOLD: f64 = 0.1;
@@ -182,6 +183,7 @@ pub fn build_parquet_writer_properties(
                 };
                 if should_apply_int32_delta(stats, ndv, num_rows) {
                     if let Some(path) = column_paths.get(&column_id) {
+                        info!("applying DBP to column {}", path.display());
                         builder = builder
                             .set_column_dictionary_enabled(path.clone(), false)
                             .set_column_encoding(path.clone(), Encoding::DELTA_BINARY_PACKED);

src/query/storages/common/blocks/src/parquet_rs.rs

@@ -31,10 +31,14 @@ use databend_common_expression::ColumnId;
 use databend_common_expression::ComputedExpr;
 use databend_common_expression::DataBlock;
 use databend_common_expression::FieldIndex;
+use databend_common_expression::Scalar;
+use databend_common_expression::TableDataType;
 use databend_common_expression::TableField;
 use databend_common_expression::TableSchema;
 use databend_common_expression::TableSchemaRef;
 use databend_common_expression::types::DataType;
+use databend_common_expression::types::number::NumberScalar;
+use databend_common_expression::types::NumberDataType;
 use databend_common_io::constants::DEFAULT_BLOCK_BUFFER_SIZE;
 use databend_common_meta_app::schema::TableIndex;
 use databend_common_native::write::NativeWriter;
@@ -407,6 +411,12 @@ impl StreamBlockBuilder {
             let mut cols_ndv = self.column_stats_state.peek_cols_ndv();
             cols_ndv.extend(self.block_stats_builder.peek_cols_ndv());
             let cols_stats = self.column_stats_state.peek_column_stats()?;
+            log_first_chunk_delta_inputs(
+                &self.properties,
+                block.num_rows(),
+                &cols_ndv,
+                &cols_stats,
+            );
             self.block_writer
                 .start(ColumnsNdvInfo::new(block.num_rows(), cols_ndv, cols_stats))?;
         }
@@ -641,3 +651,89 @@ impl StreamBlockProperties {
         }))
     }
 }
+
+fn log_first_chunk_delta_inputs(
+    properties: &StreamBlockProperties,
+    num_rows: usize,
+    cols_ndv: &HashMap<ColumnId, usize>,
+    cols_stats: &StatisticsOfColumns,
+) {
+    if !properties
+        .write_settings
+        .enable_parquet_int32_delta_encoding
+        || num_rows == 0
+        || !log::log_enabled!(log::Level::Info)
+    {
+        return;
+    }
+
+    let mut entries = Vec::new();
+    for field in properties.source_schema.leaf_fields() {
+        if !matches!(
+            field.data_type().remove_nullable(),
+            TableDataType::Number(NumberDataType::Int32)
+        ) {
+            continue;
+        }
+
+        let column_id = field.column_id();
+        let Some(ndv) = cols_ndv.get(&column_id).copied() else {
+            continue;
+        };
+        let Some(stats) = cols_stats.get(&column_id) else {
+            continue;
+        };
+
+        let ndv_ratio = (ndv as f64) / (num_rows as f64);
+        let mut entry = format!(
+            "{}: rows={} first_chunk_ndv={} ndv_ratio={:.4}",
+            field.name(),
+            num_rows,
+            ndv,
+            ndv_ratio
+        );
+
+        if let (Some(min), Some(max)) = (
+            scalar_to_i64(&stats.min),
+            scalar_to_i64(&stats.max),
+        ) {
+            if max >= min && ndv > 0 {
+                let span = max - min + 1;
+                let contiguous_ratio = span as f64 / (ndv as f64);
+                entry.push_str(&format!(
+                    " min={} max={} span={} contiguous_ratio={:.4}",
+                    min, max, span, contiguous_ratio
+                ));
+            }
+        } else {
+            entry.push_str(&format!(" min={:?} max={:?}", stats.min, stats.max));
+        }
+
+        entries.push(entry);
+    }
+
+    if !entries.is_empty() {
+        log::info!(
+            "parquet delta heuristics (first chunk) rows={} columns=[{}]",
+            num_rows,
+            entries.join("; ")
+        );
+    }
+}
+
+fn scalar_to_i64(val: &Scalar) -> Option<i64> {
+    match val {
+        Scalar::Number(num) => match num {
+            NumberScalar::Int8(v) => Some(*v as i64),
+            NumberScalar::Int16(v) => Some(*v as i64),
+            NumberScalar::Int32(v) => Some(*v as i64),
+            NumberScalar::Int64(v) => Some(*v),
+            NumberScalar::UInt8(v) => Some(*v as i64),
+            NumberScalar::UInt16(v) => Some(*v as i64),
+            NumberScalar::UInt32(v) => Some(*v as i64),
+            NumberScalar::UInt64(v) => i64::try_from(*v).ok(),
+            _ => None,
+        },
+        _ => None,
+    }
+}