feat: support real system func pg_is_visible

proto/expr.proto

@@ -328,6 +328,7 @@ message ExprNode {
     PG_IS_IN_RECOVERY = 2411;
     RW_RECOVERY_STATUS = 2412;
     RW_EPOCH_TO_TS = 2413;
+    PG_TABLE_IS_VISIBLE = 2414;
 
     // EXTERNAL
     ICEBERG_TRANSFORM = 2201;

src/frontend/src/binder/expr/function/builtin_scalar.rs

@@ -598,7 +598,7 @@ impl Binder {
                     Ok(ExprImpl::literal_varchar(new_value.to_string()))
                 }))),
                 ("format_type", raw_call(ExprType::FormatType)),
-                ("pg_table_is_visible", raw_literal(ExprImpl::literal_bool(true))),
+                ("pg_table_is_visible", raw_call(ExprType::PgTableIsVisible)),
                 ("pg_type_is_visible", raw_literal(ExprImpl::literal_bool(true))),
                 ("pg_get_constraintdef", raw_literal(ExprImpl::literal_null(DataType::Varchar))),
                 ("pg_get_partkeydef", raw_literal(ExprImpl::literal_null(DataType::Varchar))),

src/frontend/src/catalog/schema_catalog.rs

@@ -824,6 +824,17 @@ impl SchemaCatalog {
         }
     }
 
+    pub fn contains_object(&self, oid: u32) -> bool {
+        self.table_by_id.contains_key(&TableId::new(oid))
+            || self.index_by_id.contains_key(&IndexId::new(oid))
+            || self.source_by_id.contains_key(&oid)
+            || self.sink_by_id.contains_key(&oid)
+            || self.view_by_id.contains_key(&oid)
+            || self.function_by_id.contains_key(&FunctionId::new(oid))
+            || self.subscription_by_id.contains_key(&oid)
+            || self.connection_by_id.contains_key(&oid)
+    }
+
     pub fn id(&self) -> SchemaId {
         self.id
     }

src/frontend/src/catalog/system_catalog/mod.rs

@@ -314,6 +314,10 @@ pub fn get_sys_views_in_schema(schema_name: &str) -> Vec<Arc<ViewCatalog>> {
         .collect()
 }
 
+pub fn is_system_catalog(oid: u32) -> bool {
+    oid >= SYS_CATALOG_START_ID as u32
+}
+
 /// The global registry of all builtin catalogs.
 pub static SYS_CATALOGS: LazyLock<SystemCatalog> = LazyLock::new(|| {
     tracing::info!("found {} catalogs", SYS_CATALOGS_SLICE.len());

src/frontend/src/expr/function_impl/mod.rs

@@ -22,5 +22,6 @@ mod pg_get_viewdef;
 mod pg_index_column_has_property;
 mod pg_indexes_size;
 mod pg_relation_size;
+mod pg_table_is_visible;
 mod rw_epoch_to_ts;
 mod rw_recovery_status;

src/frontend/src/expr/function_impl/pg_table_is_visible.rs

@@ -0,0 +1,72 @@
+// Copyright 2025 RisingWave 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 risingwave_common::acl::AclMode;
+use risingwave_common::session_config::SearchPath;
+use risingwave_expr::{capture_context, function, Result};
+use risingwave_pb::user::grant_privilege::Object as GrantObject;
+
+use super::context::{AUTH_CONTEXT, CATALOG_READER, DB_NAME, SEARCH_PATH, USER_INFO_READER};
+use crate::catalog::system_catalog::is_system_catalog;
+use crate::catalog::CatalogReader;
+use crate::expr::function_impl::has_privilege::user_not_found_err;
+use crate::session::AuthContext;
+use crate::user::user_service::UserInfoReader;
+
+#[function("pg_table_is_visible(int4) -> boolean")]
+fn pg_table_is_visible(oid: i32) -> Result<Option<bool>> {
+    pg_table_is_visible_impl_captured(oid)
+}
+
+#[capture_context(CATALOG_READER, USER_INFO_READER, AUTH_CONTEXT, SEARCH_PATH, DB_NAME)]
+fn pg_table_is_visible_impl(
+    catalog: &CatalogReader,
+    user_info: &UserInfoReader,
+    auth_context: &AuthContext,
+    search_path: &SearchPath,
+    db_name: &str,
+    oid: i32,
+) -> Result<Option<bool>> {
+    // To maintain consistency with PostgreSQL, we ensure that system catalogs are always visible.
+    if is_system_catalog(oid as u32) {
+        return Ok(Some(true));
+    }
+
+    let catalog_reader = catalog.read_guard();
+    let user_reader = user_info.read_guard();
+    let user_info = user_reader
+        .get_user_by_name(&auth_context.user_name)
+        .ok_or(user_not_found_err(
+            format!("User {} not found", auth_context.user_name).as_str(),
+        ))?;
+    // Return true only if:
+    // 1. The schema of the object exists in the search path.
+    // 2. User have `USAGE` privilege on the schema.
+    for schema in search_path.path() {
+        if let Ok(schema) = catalog_reader.get_schema_by_name(db_name, schema) {
+            if schema.contains_object(oid as u32) {
+                return if user_info.is_super
+                    || user_info
+                        .check_privilege(&GrantObject::SchemaId(schema.id()), AclMode::Usage)
+                {
+                    Ok(Some(true))
+                } else {
+                    Ok(Some(false))
+                };
+            }
+        }
+    }
+
+    Ok(None)
+}

src/frontend/src/expr/pure.rs

@@ -294,7 +294,8 @@ impl ExprVisitor for ImpureAnalyzer {
             | Type::HasSchemaPrivilege
             | Type::MakeTimestamptz
             | Type::PgIsInRecovery
-            | Type::RwRecoveryStatus => self.impure = true,
+            | Type::RwRecoveryStatus
+            | Type::PgTableIsVisible => self.impure = true,
         }
     }
 }

src/frontend/src/optimizer/plan_expr_visitor/strong.rs

@@ -322,6 +322,7 @@ impl Strong {
             | ExprType::PgGetSerialSequence
             | ExprType::PgIndexColumnHasProperty
             | ExprType::PgIsInRecovery
+            | ExprType::PgTableIsVisible
             | ExprType::RwRecoveryStatus
             | ExprType::IcebergTransform
             | ExprType::HasTablePrivilege