feat: Add disable_list_batch for webhdfs (#17345)

* feat: Add disable_list_batch for webhdfs

Signed-off-by: Xuanwo <[email protected]>

* Fix test

Signed-off-by: Xuanwo <[email protected]>

---------

Signed-off-by: Xuanwo <[email protected]>

Author: Xuanwo

src/common/storage/src/operator.rs

@@ -355,11 +355,15 @@ fn init_moka_operator(v: &StorageMokaConfig) -> Result<impl Builder> {
 
 /// init_webhdfs_operator will init a WebHDFS operator
 fn init_webhdfs_operator(v: &StorageWebhdfsConfig) -> Result<impl Builder> {
-    let builder = services::Webhdfs::default()
+    let mut builder = services::Webhdfs::default()
         .endpoint(&v.endpoint_url)
         .root(&v.root)
         .delegation(&v.delegation);
 
+    if v.disable_list_batch {
+        builder = builder.disable_list_batch();
+    }
+
     Ok(builder)
 }
 

src/meta/app/src/storage/storage_params.rs

@@ -522,14 +522,16 @@ pub struct StorageWebhdfsConfig {
     pub endpoint_url: String,
     pub root: String,
     pub delegation: String,
+    pub disable_list_batch: bool,
 }
 
 impl Debug for StorageWebhdfsConfig {
     fn fmt(&self, f: &mut Formatter) -> std::fmt::Result {
         let mut ds = f.debug_struct("StorageWebhdfsConfig");
 
         ds.field("endpoint_url", &self.endpoint_url)
-            .field("root", &self.root);
+            .field("root", &self.root)
+            .field("disable_list_batch", &self.disable_list_batch);
 
         ds.field("delegation", &mask_string(&self.delegation, 3));
 

src/meta/proto-conv/src/config_from_to_protobuf_impl.rs

@@ -261,6 +261,7 @@ impl FromToProto for StorageWebhdfsConfig {
             endpoint_url: p.endpoint_url,
             root: p.root,
             delegation: p.delegation,
+            disable_list_batch: p.disable_list_batch,
         })
     }
 
@@ -271,6 +272,7 @@ impl FromToProto for StorageWebhdfsConfig {
             endpoint_url: self.endpoint_url.clone(),
             root: self.root.clone(),
             delegation: self.delegation.clone(),
+            disable_list_batch: self.disable_list_batch,
 
             username: String::new(), // reserved for future use
             password: String::new(), // reserved for future use

src/meta/proto-conv/src/util.rs

@@ -146,6 +146,7 @@ const META_CHANGE_LOG: &[(u64, &str)] = &[
     (114, "2024-12-12: Add: New DataType Interval."),
     (115, "2024-12-16: Add: udf.proto: add UDAFScript and UDAFServer"),
     (116, "2025-01-09: Add: MarkedDeletedIndexMeta"),
+    (117, "2025-01-21: Add: config.proto: add disable_list_batch in WebhdfsConfig")
     // Dear developer:
     //      If you're gonna add a new metadata version, you'll have to add a test for it.
     //      You could just copy an existing test file(e.g., `../tests/it/v024_table_meta.rs`)

src/meta/proto-conv/tests/it/main.rs

@@ -114,3 +114,4 @@ mod v113_warehouse_grantobject;
 mod v114_interval_datatype;
 mod v115_add_udaf_script;
 mod v116_marked_deleted_index_meta;
+mod v117_webhdfs_add_disable_list_batch;

src/meta/proto-conv/tests/it/user_proto_conv.rs

@@ -321,6 +321,7 @@ pub(crate) fn test_webhdfs_stage_info() -> mt::principal::StageInfo {
                 endpoint_url: "https://webhdfs.example.com".to_string(),
                 root: "/path/to/stage/files".to_string(),
                 delegation: "<delegation_token>".to_string(),
+                disable_list_batch: false,
             }),
         },
         is_temporary: false,

src/meta/proto-conv/tests/it/user_stage.rs

@@ -94,6 +94,7 @@ fn test_user_stage_webhdfs_v30() -> anyhow::Result<()> {
                 endpoint_url: "https://webhdfs.example.com".to_string(),
                 root: "/path/to/stage/files".to_string(),
                 delegation: "<delegation_token>".to_string(),
+                disable_list_batch: false,
             }),
         },
         file_format_params: mt::principal::FileFormatParams::Json(

src/meta/proto-conv/tests/it/v030_user_stage.rs

@@ -53,6 +53,7 @@ fn test_decode_v30_user_stage() -> anyhow::Result<()> {
                 endpoint_url: "https://webhdfs.example.com".to_string(),
                 root: "/path/to/stage/files".to_string(),
                 delegation: "<delegation_token>".to_string(),
+                disable_list_batch: false,
             }),
         },
         file_format_params: mt::principal::FileFormatParams::Json(

src/meta/proto-conv/tests/it/v031_copy_max_file.rs

@@ -51,6 +51,7 @@ fn test_decode_v31_copy_max_file() -> anyhow::Result<()> {
                 endpoint_url: "https://webhdfs.example.com".to_string(),
                 root: "/path/to/stage/files".to_string(),
                 delegation: "<delegation_token>".to_string(),
+                disable_list_batch: false,
             }),
         },
         file_format_params: mt::principal::FileFormatParams::Json(

src/meta/proto-conv/tests/it/v117_webhdfs_add_disable_list_batch.rs

@@ -0,0 +1,88 @@
+// Copyright 2023 Datafuse Labs.
+//
+// Licensed 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.
+
+use chrono::DateTime;
+use chrono::Utc;
+use databend_common_meta_app as mt;
+use databend_common_meta_app::principal::UserIdentity;
+use databend_common_meta_app::storage::StorageParams;
+use databend_common_meta_app::storage::StorageWebhdfsConfig;
+use fastrace::func_name;
+
+use crate::common;
+
+// These bytes are built when a new version in introduced,
+// and are kept for backward compatibility test.
+//
+// *************************************************************
+// * These messages should never be updated,                   *
+// * only be added when a new version is added,                *
+// * or be removed when an old version is no longer supported. *
+// *************************************************************
+//
+#[test]
+fn test_v117_webhdfs_add_disable_list_batch() -> anyhow::Result<()> {
+    // Encoded data of version 117 of databend_common_meta_app::principal::user_stage::StageInfo:
+    // It is generated with common::test_pb_from_to().
+    let stage_info_v117 = vec![
+        10, 22, 119, 101, 98, 104, 100, 102, 115, 58, 47, 47, 100, 105, 114, 47, 116, 111, 47, 102,
+        105, 108, 101, 115, 16, 1, 26, 12, 10, 10, 42, 8, 48, 1, 160, 6, 117, 168, 6, 24, 42, 11,
+        10, 2, 48, 2, 16, 142, 8, 24, 1, 56, 1, 50, 4, 116, 101, 115, 116, 56, 100, 66, 29, 10, 8,
+        100, 97, 116, 97, 98, 101, 110, 100, 18, 11, 100, 97, 116, 97, 98, 101, 110, 100, 46, 114,
+        115, 160, 6, 117, 168, 6, 24, 74, 10, 34, 8, 8, 2, 160, 6, 117, 168, 6, 24, 82, 23, 49, 57,
+        55, 48, 45, 48, 49, 45, 48, 49, 32, 48, 48, 58, 48, 48, 58, 48, 48, 32, 85, 84, 67, 160, 6,
+        117, 168, 6, 24,
+    ];
+
+    let want = || mt::principal::StageInfo {
+        stage_name: "webhdfs://dir/to/files".to_string(),
+        stage_type: mt::principal::StageType::External,
+        stage_params: mt::principal::StageParams {
+            storage: StorageParams::Webhdfs(StorageWebhdfsConfig {
+                disable_list_batch: true,
+                ..Default::default()
+            }),
+        },
+        is_temporary: false,
+        file_format_params: mt::principal::FileFormatParams::Json(
+            mt::principal::JsonFileFormatParams {
+                compression: mt::principal::StageFileCompression::Bz2,
+            },
+        ),
+        copy_options: mt::principal::CopyOptions {
+            on_error: mt::principal::OnErrorMode::AbortNum(2),
+            size_limit: 1038,
+            max_files: 0,
+            split_size: 0,
+            purge: true,
+            single: false,
+            max_file_size: 0,
+            disable_variant_check: true,
+            return_failed_only: false,
+            detailed_output: false,
+        },
+        comment: "test".to_string(),
+        number_of_files: 100,
+        creator: Some(UserIdentity {
+            username: "databend".to_string(),
+            hostname: "databend.rs".to_string(),
+        }),
+        created_on: DateTime::<Utc>::default(),
+    };
+
+    common::test_pb_from_to(func_name!(), want())?;
+    common::test_load_old(func_name!(), stage_info_v117.as_slice(), 117, want())?;
+
+    Ok(())
+}