feat: new table option FUSE_OPT_KEY_ENABLE_PARQUET_ENCODING

dantengsky · dantengsky · commit 29ccc9214d00 · 2025-05-21T15:47:47.000+08:00
When this option is set to true, fuse table with parquet storage format
will enable encoding and dictionary during serialization. The defult
encoding policy of paquet-rs will be used.
diff --git a/src/query/formats/src/output_format/parquet.rs b/src/query/formats/src/output_format/parquet.rs
@@ -53,7 +53,8 @@ impl OutputFormat for ParquetOutputFormat {
             return Ok(vec![]);
         }
         let mut buf = Vec::with_capacity(DEFAULT_BLOCK_BUFFER_SIZE);
-        let _ = blocks_to_parquet(&self.schema, blocks, &mut buf, TableCompression::Zstd)?;
+        // Unloading data, enable encoding unconditionally in this case, since ...
+        let _ = blocks_to_parquet(&self.schema, blocks, &mut buf, TableCompression::Zstd, true)?;
         Ok(buf)
     }
 }
diff --git a/src/query/service/src/interpreters/common/table_option_validation.rs b/src/query/service/src/interpreters/common/table_option_validation.rs
@@ -30,6 +30,7 @@ use databend_common_storages_fuse::FUSE_OPT_KEY_BLOCK_PER_SEGMENT;
 use databend_common_storages_fuse::FUSE_OPT_KEY_DATA_RETENTION_NUM_SNAPSHOTS_TO_KEEP;
 use databend_common_storages_fuse::FUSE_OPT_KEY_DATA_RETENTION_PERIOD_IN_HOURS;
 use databend_common_storages_fuse::FUSE_OPT_KEY_ENABLE_AUTO_VACUUM;
+use databend_common_storages_fuse::FUSE_OPT_KEY_ENABLE_PARQUET_ENCODING;
 use databend_common_storages_fuse::FUSE_OPT_KEY_FILE_SIZE;
 use databend_common_storages_fuse::FUSE_OPT_KEY_ROW_AVG_DEPTH_THRESHOLD;
 use databend_common_storages_fuse::FUSE_OPT_KEY_ROW_PER_BLOCK;
@@ -83,6 +84,7 @@ pub static CREATE_FUSE_OPTIONS: LazyLock<HashSet<&'static str>> = LazyLock::new(
     r.insert(OPT_KEY_TEMP_PREFIX);
     r.insert(OPT_KEY_SEGMENT_FORMAT);
     r.insert(OPT_KEY_ENABLE_COPY_DEDUP_FULL_PATH);
+    r.insert(FUSE_OPT_KEY_ENABLE_PARQUET_ENCODING);
     r
 });
 
@@ -243,3 +245,19 @@ where
     }
     Ok(())
 }
+
+pub fn is_valid_fuse_parquet_encoding_opt(
+    options: &BTreeMap<String, String>,
+) -> databend_common_exception::Result<()> {
+    is_valid_bool_opt(FUSE_OPT_KEY_ENABLE_PARQUET_ENCODING, options)
+}
+
+fn is_valid_bool_opt(
+    key: &str,
+    options: &BTreeMap<String, String>,
+) -> databend_common_exception::Result<()> {
+    if let Some(value) = options.get(key) {
+        value.parse::<bool>()?;
+    }
+    Ok(())
+}
diff --git a/src/query/service/src/interpreters/interpreter_table_create.rs b/src/query/service/src/interpreters/interpreter_table_create.rs
@@ -70,6 +70,7 @@ use crate::interpreters::common::table_option_validation::is_valid_bloom_index_c
 use crate::interpreters::common::table_option_validation::is_valid_change_tracking;
 use crate::interpreters::common::table_option_validation::is_valid_create_opt;
 use crate::interpreters::common::table_option_validation::is_valid_data_retention_period;
+use crate::interpreters::common::table_option_validation::is_valid_fuse_parquet_encoding_opt;
 use crate::interpreters::common::table_option_validation::is_valid_option_of_type;
 use crate::interpreters::common::table_option_validation::is_valid_random_seed;
 use crate::interpreters::common::table_option_validation::is_valid_row_per_block;
@@ -459,6 +460,8 @@ impl CreateTableInterpreter {
         is_valid_random_seed(&table_meta.options)?;
         // check table level data_retention_period_in_hours
         is_valid_data_retention_period(&table_meta.options)?;
+        // check enable_parquet_encoding
+        is_valid_fuse_parquet_encoding_opt(&table_meta.options)?;
 
         // Same as settings of FUSE_OPT_KEY_ENABLE_AUTO_VACUUM, expect value type is unsigned integer
         is_valid_option_of_type::<u32>(&table_meta.options, FUSE_OPT_KEY_ENABLE_AUTO_VACUUM)?;
diff --git a/src/query/service/src/interpreters/interpreter_table_set_options.rs b/src/query/service/src/interpreters/interpreter_table_set_options.rs
@@ -53,6 +53,7 @@ use crate::interpreters::common::table_option_validation::is_valid_block_per_seg
 use crate::interpreters::common::table_option_validation::is_valid_bloom_index_columns;
 use crate::interpreters::common::table_option_validation::is_valid_create_opt;
 use crate::interpreters::common::table_option_validation::is_valid_data_retention_period;
+use crate::interpreters::common::table_option_validation::is_valid_fuse_parquet_encoding_opt;
 use crate::interpreters::common::table_option_validation::is_valid_option_of_type;
 use crate::interpreters::common::table_option_validation::is_valid_row_per_block;
 use crate::interpreters::Interpreter;
@@ -90,6 +91,8 @@ impl Interpreter for SetOptionsInterpreter {
         is_valid_row_per_block(&self.plan.set_options)?;
         // check data_retention_period
         is_valid_data_retention_period(&self.plan.set_options)?;
+        // check enable_parquet_encoding
+        is_valid_fuse_parquet_encoding_opt(&self.plan.set_options)?;
 
         // check storage_format
         let error_str = "invalid opt for fuse table in alter table statement";
diff --git a/src/query/service/src/test_kits/block_writer.rs b/src/query/service/src/test_kits/block_writer.rs
@@ -144,6 +144,7 @@ impl<'a> BlockWriter<'a> {
                 vec![index_block],
                 &mut data,
                 TableCompression::None,
+                false,
             )?;
             let size = data.len() as u64;
             data_accessor.write(&location.0, data).await?;
diff --git a/src/query/storages/common/blocks/src/parquet_rs.rs b/src/query/storages/common/blocks/src/parquet_rs.rs
@@ -22,6 +22,7 @@ use parquet::arrow::ArrowWriter;
 use parquet::basic::Encoding;
 use parquet::file::properties::EnabledStatistics;
 use parquet::file::properties::WriterProperties;
+use parquet::file::properties::WriterVersion;
 use parquet::format::FileMetaData;
 
 /// Serialize data blocks to parquet format.
@@ -30,17 +31,41 @@ pub fn blocks_to_parquet(
     blocks: Vec<DataBlock>,
     write_buffer: &mut Vec<u8>,
     compression: TableCompression,
+    enable_encoding: bool,
 ) -> Result<FileMetaData> {
     assert!(!blocks.is_empty());
-    let props = WriterProperties::builder()
+    let builder = WriterProperties::builder()
         .set_compression(compression.into())
         // use `usize::MAX` to effectively limit the number of row groups to 1
         .set_max_row_group_size(usize::MAX)
-        .set_encoding(Encoding::PLAIN)
-        .set_dictionary_enabled(false)
         .set_statistics_enabled(EnabledStatistics::None)
-        .set_bloom_filter_enabled(false)
-        .build();
+        .set_bloom_filter_enabled(false);
+
+    let builder = if enable_encoding {
+        // Enable dictionary encoding and fallback encodings.
+        //
+        // Memo for quick lookup:
+        // The fallback encoding "strategy" used by parquet-54.2.1 is:
+        //
+        // ~~~
+        //   (Type::BOOLEAN, WriterVersion::PARQUET_2_0) => Encoding::RLE,
+        //   (Type::INT32, WriterVersion::PARQUET_2_0) => Encoding::DELTA_BINARY_PACKED,
+        //   (Type::INT64, WriterVersion::PARQUET_2_0) => Encoding::DELTA_BINARY_PACKED,
+        //   (Type::BYTE_ARRAY, WriterVersion::PARQUET_2_0) => Encoding::DELTA_BYTE_ARRAY,
+        //   (Type::FIXED_LEN_BYTE_ARRAY, WriterVersion::PARQUET_2_0) => Encoding::DELTA_BYTE_ARRAY,
+        //   _ => Encoding::PLAIN,
+        // ~~~
+        //
+        builder
+            .set_writer_version(WriterVersion::PARQUET_2_0)
+            .set_dictionary_enabled(true)
+    } else {
+        builder
+            .set_dictionary_enabled(false)
+            .set_encoding(Encoding::PLAIN)
+    };
+
+    let props = builder.build();
     let batches = blocks
         .into_iter()
         .map(|block| block.to_record_batch(table_schema))
diff --git a/src/query/storages/fuse/src/constants.rs b/src/query/storages/fuse/src/constants.rs
@@ -18,12 +18,12 @@ pub const FUSE_OPT_KEY_ROW_PER_BLOCK: &str = "row_per_block";
 pub const FUSE_OPT_KEY_ROW_PER_PAGE: &str = "row_per_page";
 pub const FUSE_OPT_KEY_ROW_AVG_DEPTH_THRESHOLD: &str = "row_avg_depth_threshold";
 pub const FUSE_OPT_KEY_FILE_SIZE: &str = "file_size";
-
 pub const FUSE_OPT_KEY_DATA_RETENTION_PERIOD_IN_HOURS: &str = "data_retention_period_in_hours";
 pub const FUSE_OPT_KEY_DATA_RETENTION_NUM_SNAPSHOTS_TO_KEEP: &str =
     "data_retention_num_snapshots_to_keep";
 pub const FUSE_OPT_KEY_ENABLE_AUTO_VACUUM: &str = "enable_auto_vacuum";
 pub const FUSE_OPT_KEY_ATTACH_COLUMN_IDS: &str = "attach_column_ids";
+pub const FUSE_OPT_KEY_ENABLE_PARQUET_ENCODING: &str = "enable_parquet_encoding";
 
 pub const FUSE_TBL_BLOCK_PREFIX: &str = "_b";
 pub const FUSE_TBL_BLOCK_INDEX_PREFIX: &str = "_i";
diff --git a/src/query/storages/fuse/src/fuse_table.rs b/src/query/storages/fuse/src/fuse_table.rs
@@ -126,6 +126,7 @@ use crate::FUSE_OPT_KEY_BLOCK_IN_MEM_SIZE_THRESHOLD;
 use crate::FUSE_OPT_KEY_BLOCK_PER_SEGMENT;
 use crate::FUSE_OPT_KEY_DATA_RETENTION_NUM_SNAPSHOTS_TO_KEEP;
 use crate::FUSE_OPT_KEY_DATA_RETENTION_PERIOD_IN_HOURS;
+use crate::FUSE_OPT_KEY_ENABLE_PARQUET_ENCODING;
 use crate::FUSE_OPT_KEY_FILE_SIZE;
 use crate::FUSE_OPT_KEY_ROW_PER_BLOCK;
 use crate::FUSE_OPT_KEY_ROW_PER_PAGE;
@@ -277,11 +278,14 @@ impl FuseTable {
         let block_per_seg =
             self.get_option(FUSE_OPT_KEY_BLOCK_PER_SEGMENT, DEFAULT_BLOCK_PER_SEGMENT);
 
+        let enable_parquet_encoding = self.get_option(FUSE_OPT_KEY_ENABLE_PARQUET_ENCODING, false);
+
         WriteSettings {
             storage_format: self.storage_format,
             table_compression: self.table_compression,
             max_page_size,
             block_per_seg,
+            enable_parquet_encoding,
         }
     }
 
diff --git a/src/query/storages/fuse/src/io/write/block_writer.rs b/src/query/storages/fuse/src/io/write/block_writer.rs
@@ -69,8 +69,13 @@ pub fn serialize_block(
     let schema = Arc::new(schema.remove_virtual_computed_fields());
     match write_settings.storage_format {
         FuseStorageFormat::Parquet => {
-            let result =
-                blocks_to_parquet(&schema, vec![block], buf, write_settings.table_compression)?;
+            let result = blocks_to_parquet(
+                &schema,
+                vec![block],
+                buf,
+                write_settings.table_compression,
+                write_settings.enable_parquet_encoding,
+            )?;
             let meta = column_parquet_metas(&result, &schema)?;
             Ok(meta)
         }
diff --git a/src/query/storages/fuse/src/io/write/write_settings.rs b/src/query/storages/fuse/src/io/write/write_settings.rs
@@ -28,6 +28,7 @@ pub struct WriteSettings {
     pub max_page_size: usize,
 
     pub block_per_seg: usize,
+    pub enable_parquet_encoding: bool,
 }
 
 impl Default for WriteSettings {
@@ -37,6 +38,7 @@ impl Default for WriteSettings {
             table_compression: TableCompression::default(),
             max_page_size: DEFAULT_ROW_PER_PAGE,
             block_per_seg: DEFAULT_BLOCK_PER_SEGMENT,
+            enable_parquet_encoding: false,
         }
     }
 }
diff --git a/src/query/storages/result_cache/src/write/writer.rs b/src/query/storages/result_cache/src/write/writer.rs
@@ -72,11 +72,13 @@ impl ResultCacheWriter {
     #[async_backtrace::framed]
     pub async fn write_to_storage(&self) -> Result<String> {
         let mut buf = Vec::with_capacity(self.current_bytes);
+        // TODO doc why encoding is not enabled
         let _ = blocks_to_parquet(
             &self.schema,
             self.blocks.clone(),
             &mut buf,
             TableCompression::None,
+            false,
         )?;
 
         let file_location = format!("{}/{}.parquet", self.location, Uuid::new_v4().as_simple());
diff --git a/tests/sqllogictests/suites/base/09_fuse_engine/09_0046_parquet_encoding.test b/tests/sqllogictests/suites/base/09_fuse_engine/09_0046_parquet_encoding.test
@@ -0,0 +1,29 @@
+statement ok
+create or replace database test_tbl_opt_parquet_encoding;
+
+statement ok
+use test_tbl_opt_parquet_encoding;
+
+# Test create table with parquet encoding option
+
+statement ok
+create or replace table t_encoded (c int, s string) enable_parquet_encoding = 'true' compression = 'lz4';
+
+statement ok
+create or replace table t(c int, s string) compression = 'lz4';
+
+statement ok
+insert into t_encoded(c, s) select number as c, to_string(number) as s from numbers(500000);
+
+statement ok
+insert into t(c, s) select number as c, to_string(number) as s from numbers(500000);
+
+query T
+select humanize_size(file_size) from fuse_block('test_tbl_opt_parquet_encoding', 't_encoded');
+----
+2.30 MiB
+
+query T
+select humanize_size(file_size) from fuse_block('test_tbl_opt_parquet_encoding', 't');
+----
+3.91 MiB

Original file line number	Diff line number	Diff line change
`@@ -53,7 +53,8 @@ impl OutputFormat for ParquetOutputFormat {`
`53`	`53`	`return Ok(vec![]);`
`54`	`54`	`}`
`55`	`55`	`let mut buf = Vec::with_capacity(DEFAULT_BLOCK_BUFFER_SIZE);`
`56`		`- let _ = blocks_to_parquet(&self.schema, blocks, &mut buf, TableCompression::Zstd)?;`
	`56`	`+ // Unloading data, enable encoding unconditionally in this case, since ...`
	`57`	`+ let _ = blocks_to_parquet(&self.schema, blocks, &mut buf, TableCompression::Zstd, true)?;`
`57`	`58`	`Ok(buf)`
`58`	`59`	`}`
`59`	`60`	`}`
Original file line number	Diff line number	Diff line change
`@@ -28,6 +28,7 @@ pub struct WriteSettings {`
`28`	`28`	`pub max_page_size: usize,`
`29`	`29`
`30`	`30`	`pub block_per_seg: usize,`
	`31`	`+ pub enable_parquet_encoding: bool,`
`31`	`32`	`}`
`32`	`33`
`33`	`34`	`impl Default for WriteSettings {`
`@@ -37,6 +38,7 @@ impl Default for WriteSettings {`
`37`	`38`	`table_compression: TableCompression::default(),`
`38`	`39`	`max_page_size: DEFAULT_ROW_PER_PAGE,`
`39`	`40`	`block_per_seg: DEFAULT_BLOCK_PER_SEGMENT,`
	`41`	`+ enable_parquet_encoding: false,`
`40`	`42`	`}`
`41`	`43`	`}`
`42`	`44`	`}`