feat(query): add iceberg table functions (#17626)

* feat(query): add iceberg table functions

* feat(query): add iceberg table functionsm

* feat(query): add iceberg table functions

* feat(query): add iceberg table functions

Author: sundy-li

Cargo.lock

@@ -12,10 +12,12 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use std::cmp::Ordering;
 use std::collections::HashMap;
 
 use databend_common_exception::ErrorCode;
 use databend_common_exception::Result;
+use databend_common_expression::types::NumberScalar;
 use databend_common_expression::Scalar;
 use log::debug;
 
@@ -99,3 +101,108 @@ impl TableArgs {
         }
     }
 }
+
+pub fn string_value(value: &Scalar) -> Result<String> {
+    match value {
+        Scalar::String(val) => Ok(val.clone()),
+        _ => Err(ErrorCode::BadArguments(format!(
+            "invalid value {value} expect to be string literal."
+        ))),
+    }
+}
+
+pub fn bool_value(value: &Scalar) -> Result<bool> {
+    match value {
+        Scalar::Boolean(val) => Ok(*val),
+        _ => Err(ErrorCode::BadArguments(format!(
+            "invalid value {value} expect to be boolean literal."
+        ))),
+    }
+}
+
+pub fn string_literal(val: &str) -> Scalar {
+    Scalar::String(val.to_string())
+}
+
+pub fn bool_literal(val: bool) -> Scalar {
+    Scalar::Boolean(val)
+}
+
+pub fn u64_literal(val: u64) -> Scalar {
+    Scalar::Number(NumberScalar::UInt64(val))
+}
+
+pub fn cmp_with_null(v1: &Scalar, v2: &Scalar) -> Ordering {
+    match (v1.is_null(), v2.is_null()) {
+        (true, true) => Ordering::Equal,
+        (true, false) => Ordering::Greater,
+        (false, true) => Ordering::Less,
+        (false, false) => v1.cmp(v2),
+    }
+}
+
+pub fn parse_sequence_args(table_args: &TableArgs, func_name: &str) -> Result<String> {
+    let args = table_args.expect_all_positioned(func_name, Some(1))?;
+    let sequence = string_value(&args[0])?;
+    Ok(sequence)
+}
+
+pub fn parse_db_tb_args(table_args: &TableArgs, func_name: &str) -> Result<(String, String)> {
+    let args = table_args.expect_all_positioned(func_name, Some(2))?;
+    let db = string_value(&args[0])?;
+    let tbl = string_value(&args[1])?;
+    Ok((db, tbl))
+}
+
+pub fn parse_db_tb_opt_args(
+    table_args: &TableArgs,
+    func_name: &str,
+) -> Result<(String, String, Option<String>)> {
+    let args = table_args.expect_all_positioned(func_name, None)?;
+    match args.len() {
+        3 => {
+            let arg1 = string_value(&args[0])?;
+            let arg2 = string_value(&args[1])?;
+            let arg3 = string_value(&args[2])?;
+            Ok((arg1, arg2, Some(arg3)))
+        }
+        2 => {
+            let arg1 = string_value(&args[0])?;
+            let arg2 = string_value(&args[1])?;
+            Ok((arg1, arg2, None))
+        }
+        _ => Err(ErrorCode::BadArguments(format!(
+            "expecting <database>, <table_name> and <opt_arg> (as string literals), but got {:?}",
+            args
+        ))),
+    }
+}
+
+pub fn parse_opt_opt_args(
+    table_args: &TableArgs,
+    func_name: &str,
+) -> Result<(Option<String>, Option<String>)> {
+    let args = table_args.expect_all_positioned(func_name, None)?;
+    match args.len() {
+        2 => {
+            let arg1 = string_value(&args[0])?;
+            let arg2 = string_value(&args[1])?;
+            Ok((Some(arg1), Some(arg2)))
+        }
+        1 => {
+            let arg1 = string_value(&args[0])?;
+            Ok((Some(arg1), None))
+        }
+        0 => Ok((None, None)),
+        _ => Err(ErrorCode::BadArguments(format!(
+            "expecting <opt_arg1> and <opt_arg2> (as string literals), but got {:?}",
+            args
+        ))),
+    }
+}
+
+pub fn parse_db_tb_col_args(table_args: &TableArgs, func_name: &str) -> Result<String> {
+    let args = table_args.expect_all_positioned(func_name, Some(1))?;
+    let db = string_value(&args[0])?;
+    Ok(db)
+}

src/query/catalog/src/table_args.rs

@@ -32,6 +32,7 @@ use databend_common_storages_fuse::table_functions::FuseVacuumTemporaryTable;
 use databend_common_storages_fuse::table_functions::HilbertClusteringInfoFunc;
 use databend_common_storages_fuse::table_functions::SetCacheCapacity;
 use databend_common_storages_fuse::table_functions::TableFunctionTemplate;
+use databend_common_storages_iceberg::IcebergInspectTable;
 use databend_common_storages_stream::stream_status_table_func::StreamStatusTable;
 use databend_storages_common_table_meta::table_id_ranges::SYS_TBL_FUC_ID_END;
 use databend_storages_common_table_meta::table_id_ranges::SYS_TBL_FUNC_ID_BEGIN;
@@ -354,6 +355,16 @@ impl TableFunctionFactory {
             (next_id(), Arc::new(ShowRoles::create)),
         );
 
+        creators.insert(
+            "iceberg_snapshot".to_string(),
+            (next_id(), Arc::new(IcebergInspectTable::create)),
+        );
+
+        creators.insert(
+            "iceberg_manifest".to_string(),
+            (next_id(), Arc::new(IcebergInspectTable::create)),
+        );
+
         TableFunctionFactory {
             creators: RwLock::new(creators),
         }

src/query/service/src/table_functions/table_function_factory.rs

@@ -236,8 +236,12 @@ impl Operator for Scan {
             }
             if let Some(col_stat) = v.clone() {
                 // Safe to unwrap: min, max are all `Some(_)`.
-                let min = col_stat.min.unwrap();
-                let max = col_stat.max.unwrap();
+                let Some(min) = col_stat.min.clone() else {
+                    continue;
+                };
+                let Some(max) = col_stat.max.clone() else {
+                    continue;
+                };
                 // ndv could be `None`, we will use `num_rows - null_count` as ndv instead.
                 //
                 // NOTE: don't touch the original num_rows, since it will be used in other places.

src/query/sql/src/planner/plans/scan.rs

@@ -18,6 +18,7 @@ use chrono::Utc;
 use databend_common_catalog::catalog::CATALOG_DEFAULT;
 use databend_common_catalog::plan::DataSourcePlan;
 use databend_common_catalog::table::Table;
+use databend_common_catalog::table_args::string_literal;
 use databend_common_catalog::table_args::TableArgs;
 use databend_common_catalog::table_context::TableContext;
 use databend_common_exception::ErrorCode;
@@ -40,7 +41,6 @@ use databend_storages_common_table_meta::table::OPT_KEY_CLUSTER_TYPE;
 
 use crate::io::SegmentsIO;
 use crate::table_functions::parse_db_tb_args;
-use crate::table_functions::string_literal;
 use crate::table_functions::SimpleArgFunc;
 use crate::table_functions::SimpleArgFuncTemplate;
 use crate::FuseTable;

src/query/storages/fuse/src/table_functions/hilbert_clustering_information.rs

@@ -27,13 +27,12 @@ mod fuse_vacuum_drop_aggregating_index;
 mod fuse_vacuum_drop_inverted_index;
 mod fuse_vacuum_temporary_table;
 mod hilbert_clustering_information;
-mod table_args;
 
 mod set_cache_capacity;
 
 pub use clustering_information::ClusteringInformationFunc;
 pub use clustering_statistics::ClusteringStatisticsFunc;
-use databend_common_catalog::table_args::TableArgs;
+pub use databend_common_catalog::table_args::*;
 use databend_common_catalog::table_function::TableFunction;
 pub use function_template::SimpleTableFunc;
 pub use function_template::TableFunctionTemplate;
@@ -52,4 +51,3 @@ pub use fuse_vacuum_drop_inverted_index::FuseVacuumDropInvertedIndex;
 pub use fuse_vacuum_temporary_table::FuseVacuumTemporaryTable;
 pub use hilbert_clustering_information::HilbertClusteringInfoFunc;
 pub use set_cache_capacity::SetCacheCapacity;
-pub use table_args::*;

src/query/storages/fuse/src/table_functions/mod.rs

@@ -20,6 +20,7 @@ databend-common-meta-app = { workspace = true }
 databend-common-meta-store = { workspace = true }
 databend-common-meta-types = { workspace = true }
 databend-common-pipeline-core = { workspace = true }
+databend-common-pipeline-sources = { workspace = true }
 databend-common-storage = { workspace = true }
 databend-common-storages-parquet = { workspace = true }
 databend-storages-common-table-meta = { workspace = true }