fix: change the type of oid in pg_namespace to u32 (#4541)

* fix:  change the type of oid in pg_namespace to u32

* fix: header and correct logic of update oid

Cargo.toml

@@ -155,6 +155,7 @@ rskafka = "0.5"
 rstest = "0.21"
 rstest_reuse = "0.7"
 rust_decimal = "1.33"
+rustc-hash = "2.0"
 schemars = "0.8"
 serde = { version = "1.0", features = ["derive"] }
 serde_json = { version = "1.0", features = ["float_roundtrip"] }

src/catalog/Cargo.toml

@@ -40,6 +40,7 @@ moka = { workspace = true, features = ["future", "sync"] }
 partition.workspace = true
 paste = "1.0"
 prometheus.workspace = true
+rustc-hash.workspace = true
 serde_json.workspace = true
 session.workspace = true
 snafu.workspace = true

src/catalog/src/system_schema/pg_catalog.rs

@@ -30,6 +30,7 @@ use pg_namespace::PGNamespace;
 use table::TableRef;
 pub use table_names::*;
 
+use self::pg_namespace::oid_map::{PGNamespaceOidMap, PGNamespaceOidMapRef};
 use super::memory_table::MemoryTable;
 use super::utils::tables::u32_column;
 use super::{SystemSchemaProvider, SystemSchemaProviderInner, SystemTableRef};
@@ -52,6 +53,9 @@ pub struct PGCatalogProvider {
  catalog_name: String,
  catalog_manager: Weak<dyn CatalogManager>,
  tables: HashMap<String, TableRef>,
+
+ // Workaround to store mapping of schema_name to a numeric id
+ namespace_oid_map: PGNamespaceOidMapRef,
 }
 
 impl SystemSchemaProvider for PGCatalogProvider {
@@ -85,6 +89,7 @@ impl PGCatalogProvider {
  catalog_name,
  catalog_manager,
  tables: HashMap::new(),
+ namespace_oid_map: Arc::new(PGNamespaceOidMap::new()),
  };
  provider.build_tables();
  provider
@@ -122,10 +127,12 @@ impl SystemSchemaProviderInner for PGCatalogProvider {
  table_names::PG_NAMESPACE => Some(Arc::new(PGNamespace::new(
  self.catalog_name.clone(),
  self.catalog_manager.clone(),
+ self.namespace_oid_map.clone(),
  ))),
  table_names::PG_CLASS => Some(Arc::new(PGClass::new(
  self.catalog_name.clone(),
  self.catalog_manager.clone(),
+ self.namespace_oid_map.clone(),
  ))),
  _ => None,
  }

src/catalog/src/system_schema/pg_catalog/pg_class.rs

@@ -31,6 +31,7 @@ use snafu::{OptionExt, ResultExt};
 use store_api::storage::ScanRequest;
 use table::metadata::TableType;
 
+use super::pg_namespace::oid_map::PGNamespaceOidMapRef;
 use super::{OID_COLUMN_NAME, PG_CLASS};
 use crate::error::{
  CreateRecordBatchSnafu, InternalSnafu, Result, UpgradeWeakCatalogManagerRefSnafu,
@@ -60,22 +61,30 @@ pub(super) struct PGClass {
  schema: SchemaRef,
  catalog_name: String,
  catalog_manager: Weak<dyn CatalogManager>,
+
+ // Workaround to convert schema_name to a numeric id
+ namespace_oid_map: PGNamespaceOidMapRef,
 }
 
 impl PGClass {
- pub(super) fn new(catalog_name: String, catalog_manager: Weak<dyn CatalogManager>) -> Self {
+ pub(super) fn new(
+ catalog_name: String,
+ catalog_manager: Weak<dyn CatalogManager>,
+ namespace_oid_map: PGNamespaceOidMapRef,
+ ) -> Self {
  Self {
  schema: Self::schema(),
  catalog_name,
  catalog_manager,
+ namespace_oid_map,
  }
  }
 
  fn schema() -> SchemaRef {
  Arc::new(Schema::new(vec![
  u32_column(OID_COLUMN_NAME),
  string_column(RELNAME),
- string_column(RELNAMESPACE),
+ u32_column(RELNAMESPACE),
  string_column(RELKIND),
  u32_column(RELOWNER),
  ]))
@@ -86,6 +95,7 @@ impl PGClass {
  self.schema.clone(),
  self.catalog_name.clone(),
  self.catalog_manager.clone(),
+ self.namespace_oid_map.clone(),
  )
  }
 }
@@ -155,10 +165,11 @@ struct PGClassBuilder {
  schema: SchemaRef,
  catalog_name: String,
  catalog_manager: Weak<dyn CatalogManager>,
+ namespace_oid_map: PGNamespaceOidMapRef,
 
  oid: UInt32VectorBuilder,
  relname: StringVectorBuilder,
- relnamespace: StringVectorBuilder,
+ relnamespace: UInt32VectorBuilder,
  relkind: StringVectorBuilder,
  relowner: UInt32VectorBuilder,
 }
@@ -168,15 +179,17 @@ impl PGClassBuilder {
  schema: SchemaRef,
  catalog_name: String,
  catalog_manager: Weak<dyn CatalogManager>,
+ namespace_oid_map: PGNamespaceOidMapRef,
  ) -> Self {
  Self {
  schema,
  catalog_name,
  catalog_manager,
+ namespace_oid_map,
 
  oid: UInt32VectorBuilder::with_capacity(INIT_CAPACITY),
  relname: StringVectorBuilder::with_capacity(INIT_CAPACITY),
- relnamespace: StringVectorBuilder::with_capacity(INIT_CAPACITY),
+ relnamespace: UInt32VectorBuilder::with_capacity(INIT_CAPACITY),
  relkind: StringVectorBuilder::with_capacity(INIT_CAPACITY),
  relowner: UInt32VectorBuilder::with_capacity(INIT_CAPACITY),
  }
@@ -217,6 +230,7 @@ impl PGClassBuilder {
  table: &str,
  kind: &str,
  ) {
+ let namespace_oid = self.namespace_oid_map.get_oid(schema);
  let row = [
  (OID_COLUMN_NAME, &Value::from(oid)),
  (RELNAMESPACE, &Value::from(schema)),
@@ -230,7 +244,7 @@ impl PGClassBuilder {
  }
 
  self.oid.push(Some(oid));
- self.relnamespace.push(Some(schema));
+ self.relnamespace.push(Some(namespace_oid));
  self.relname.push(Some(table));
  self.relkind.push(Some(kind));
  self.relowner.push(Some(DUMMY_OWNER_ID));

src/catalog/src/system_schema/pg_catalog/pg_namespace.rs

@@ -12,6 +12,8 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+pub(super) mod oid_map;
+
 use std::sync::{Arc, Weak};
 
 use arrow_schema::SchemaRef as ArrowSchemaRef;
@@ -25,16 +27,16 @@ use datafusion::physical_plan::streaming::PartitionStream as DfPartitionStream;
 use datatypes::scalars::ScalarVectorBuilder;
 use datatypes::schema::{Schema, SchemaRef};
 use datatypes::value::Value;
-use datatypes::vectors::{StringVectorBuilder, VectorRef};
+use datatypes::vectors::{StringVectorBuilder, UInt32VectorBuilder, VectorRef};
 use snafu::{OptionExt, ResultExt};
 use store_api::storage::ScanRequest;
 
-use super::{OID_COLUMN_NAME, PG_NAMESPACE};
+use super::{PGNamespaceOidMapRef, OID_COLUMN_NAME, PG_NAMESPACE};
 use crate::error::{
  CreateRecordBatchSnafu, InternalSnafu, Result, UpgradeWeakCatalogManagerRefSnafu,
 };
 use crate::information_schema::Predicates;
-use crate::system_schema::utils::tables::string_column;
+use crate::system_schema::utils::tables::{string_column, u32_column};
 use crate::system_schema::SystemTable;
 use crate::CatalogManager;
 
@@ -48,21 +50,29 @@ pub(super) struct PGNamespace {
  schema: SchemaRef,
  catalog_name: String,
  catalog_manager: Weak<dyn CatalogManager>,
+
+ // Workaround to convert schema_name to a numeric id
+ oid_map: PGNamespaceOidMapRef,
 }
 
 impl PGNamespace {
- pub(super) fn new(catalog_name: String, catalog_manager: Weak<dyn CatalogManager>) -> Self {
+ pub(super) fn new(
+ catalog_name: String,
+ catalog_manager: Weak<dyn CatalogManager>,
+ oid_map: PGNamespaceOidMapRef,
+ ) -> Self {
  Self {
  schema: Self::schema(),
  catalog_name,
  catalog_manager,
+ oid_map,
  }
  }
 
  fn schema() -> SchemaRef {
  Arc::new(Schema::new(vec![
  // TODO(J0HN50N133): we do not have a numeric schema id, use schema name as a workaround. Use a proper schema id once we have it.
- string_column(OID_COLUMN_NAME),
+ u32_column(OID_COLUMN_NAME),
  string_column(NSPNAME),
  ]))
  }
@@ -72,6 +82,7 @@ impl PGNamespace {
  self.schema.clone(),
  self.catalog_name.clone(),
  self.catalog_manager.clone(),
+ self.oid_map.clone(),
  )
  }
 }
@@ -138,8 +149,9 @@ struct PGNamespaceBuilder {
  schema: SchemaRef,
  catalog_name: String,
  catalog_manager: Weak<dyn CatalogManager>,
+ namespace_oid_map: PGNamespaceOidMapRef,
 
- oid: StringVectorBuilder,
+ oid: UInt32VectorBuilder,
  nspname: StringVectorBuilder,
 }
 
@@ -148,12 +160,14 @@ impl PGNamespaceBuilder {
  schema: SchemaRef,
  catalog_name: String,
  catalog_manager: Weak<dyn CatalogManager>,
+ namespace_oid_map: PGNamespaceOidMapRef,
  ) -> Self {
  Self {
  schema,
  catalog_name,
  catalog_manager,
- oid: StringVectorBuilder::with_capacity(INIT_CAPACITY),
+ namespace_oid_map,
+ oid: UInt32VectorBuilder::with_capacity(INIT_CAPACITY),
  nspname: StringVectorBuilder::with_capacity(INIT_CAPACITY),
  }
  }
@@ -178,14 +192,15 @@ impl PGNamespaceBuilder {
  }
 
  fn add_namespace(&mut self, predicates: &Predicates, schema_name: &str) {
+ let oid = self.namespace_oid_map.get_oid(schema_name);
  let row = [
- (OID_COLUMN_NAME, &Value::from(schema_name)),
+ (OID_COLUMN_NAME, &Value::from(oid)),
  (NSPNAME, &Value::from(schema_name)),
  ];
  if !predicates.eval(&row) {
  return;
  }
- self.oid.push(Some(schema_name));
+ self.oid.push(Some(oid));
  self.nspname.push(Some(schema_name));
  }
 }