feat(catalog): expose InformationSchemataBuilder as public API (#22499)

## Which issue does this PR close?

No linked issue. Happy to file one if reviewers prefer.


## Rationale for this change

Downstream catalog implementations that resolve schemas asynchronously
cannot reuse `InformationSchemaProvider` — it enumerates schemas via
`CatalogProvider::schema_names()`, which is synchronous, so an
async-only catalog has to provide its own `information_schema.schemata`
view. Today that requires either duplicating the column layout and
the row-building logic, or reaching into private items.

Exposing `InformationSchemataBuilder` and a `schemata_schema()` factory
lets external crates emit byte-for-byte-compatible `schemata` batches
without copy-pasting the contract.

## What changes are included in this PR?

- `pub fn schemata_schema() -> SchemaRef` — extracts the column-layout
  factory. `InformationSchemata::new` now calls it instead of inlining
  the schema, so there is a single source of truth.
- `InformationSchemataBuilder` becomes `pub` (was private) with a
  `Default` impl and a public `new()`. `add_schemata` and `finish` are
  bumped to `pub`. The function bodies and parameter types
  (`&str` / `Option<&str>`) are unchanged.
- `finish` now returns `Result<RecordBatch>` instead of panicking via
  an internal `.unwrap()`. The one internal caller
  (`PartitionStream::execute` for `InformationSchemata`) was previously
  wrapping `Ok(builder.finish())` and is updated to just
  `builder.finish()` since the inner expression now produces the
  `Result` directly.

## Are these changes tested?

Yes. A new unit test `schemata_builder_emits_canonical_schema_and_rows`
exercises the public API end-to-end via `Default::default()`, asserts
the produced batch's schema matches `schemata_schema()`, and verifies
the null pattern for `schema_owner`, the three `default_character_set_*`
columns, and `sql_path`. The pre-existing internal users
(`InformationSchemata::new`, `PartitionStream::execute`) continue to
exercise the same code path through the unchanged
`InformationSchemata::builder()` constructor.

## Are there any user-facing changes?

Yes — three new public items in `datafusion-catalog`: `schemata_schema`,
`InformationSchemataBuilder` (with its `new` / `add_schemata` / `finish`
methods + `Default` impl). No existing public API is broken. The
`Result<RecordBatch>` return on `finish` is a first-time-public surface,
not a regression.

Author: zfarrell

datafusion/catalog/src/information_schema.rs

@@ -967,18 +967,34 @@ struct InformationSchemata {
     config: InformationSchemaConfig,
 }
 
+/// The Arrow schema of [`information_schema.schemata`] rows.
+///
+/// Useful for downstream catalog implementations that want to declare a
+/// `TableProvider` for `schemata` before populating any rows via
+/// [`InformationSchemataBuilder`].
+///
+/// Columns and nullability match
+/// <https://www.postgresql.org/docs/current/infoschema-schemata.html>.
+///
+/// [`information_schema.schemata`]: https://www.postgresql.org/docs/current/infoschema-schemata.html
+pub fn schemata_schema() -> SchemaRef {
+    Arc::new(Schema::new(vec![
+        Field::new("catalog_name", DataType::Utf8, false),
+        Field::new("schema_name", DataType::Utf8, false),
+        Field::new("schema_owner", DataType::Utf8, true),
+        Field::new("default_character_set_catalog", DataType::Utf8, true),
+        Field::new("default_character_set_schema", DataType::Utf8, true),
+        Field::new("default_character_set_name", DataType::Utf8, true),
+        Field::new("sql_path", DataType::Utf8, true),
+    ]))
+}
+
 impl InformationSchemata {
     fn new(config: InformationSchemaConfig) -> Self {
-        let schema = Arc::new(Schema::new(vec![
-            Field::new("catalog_name", DataType::Utf8, false),
-            Field::new("schema_name", DataType::Utf8, false),
-            Field::new("schema_owner", DataType::Utf8, true),
-            Field::new("default_character_set_catalog", DataType::Utf8, true),
-            Field::new("default_character_set_schema", DataType::Utf8, true),
-            Field::new("default_character_set_name", DataType::Utf8, true),
-            Field::new("sql_path", DataType::Utf8, true),
-        ]));
-        Self { schema, config }
+        Self {
+            schema: schemata_schema(),
+            config,
+        }
     }
 
     fn builder(&self) -> InformationSchemataBuilder {
@@ -995,7 +1011,16 @@ impl InformationSchemata {
     }
 }
 
-struct InformationSchemataBuilder {
+/// Builder that produces [`RecordBatch`] values matching the schema of
+/// `information_schema.schemata` (see [`schemata_schema`]).
+///
+/// Intended for downstream catalog implementations that need to emit
+/// `schemata` rows from their own metadata source rather than going
+/// through DataFusion's `InformationSchemaProvider`, which enumerates
+/// schemas synchronously via `CatalogProviderList` and so is unsuitable
+/// for catalog backends that resolve asynchronously.
+#[derive(Debug)]
+pub struct InformationSchemataBuilder {
     schema: SchemaRef,
     catalog_name: StringBuilder,
     schema_name: StringBuilder,
@@ -1006,8 +1031,32 @@ struct InformationSchemataBuilder {
     sql_path: StringBuilder,
 }
 
+impl Default for InformationSchemataBuilder {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
 impl InformationSchemataBuilder {
-    fn add_schemata(
+    /// Construct an empty builder.
+    pub fn new() -> Self {
+        Self {
+            schema: schemata_schema(),
+            catalog_name: StringBuilder::new(),
+            schema_name: StringBuilder::new(),
+            schema_owner: StringBuilder::new(),
+            default_character_set_catalog: StringBuilder::new(),
+            default_character_set_schema: StringBuilder::new(),
+            default_character_set_name: StringBuilder::new(),
+            sql_path: StringBuilder::new(),
+        }
+    }
+
+    /// Append one row to the builder. `schema_owner` is the optional SQL
+    /// schema owner; the three `default_character_set_*` columns and
+    /// `sql_path` are written as null (DataFusion does not model those
+    /// concepts; see the PostgreSQL docs link on [`schemata_schema`]).
+    pub fn add_schemata(
         &mut self,
         catalog_name: &str,
         schema_name: &str,
@@ -1019,15 +1068,19 @@ impl InformationSchemataBuilder {
             Some(owner) => self.schema_owner.append_value(owner),
             None => self.schema_owner.append_null(),
         }
-        // refer to https://www.postgresql.org/docs/current/infoschema-schemata.html,
-        // these rows apply to a feature that is not implemented in DataFusion
         self.default_character_set_catalog.append_null();
         self.default_character_set_schema.append_null();
         self.default_character_set_name.append_null();
         self.sql_path.append_null();
     }
 
-    fn finish(&mut self) -> RecordBatch {
+    /// Finalize the builder into a [`RecordBatch`].
+    ///
+    /// Returns an error only if Arrow buffer construction fails, which
+    /// the builder's column-count and type invariants make unreachable
+    /// under normal use. The `Result` return type preserves room to add
+    /// validation in the future without a breaking API change.
+    pub fn finish(&mut self) -> Result<RecordBatch> {
         RecordBatch::try_new(
             Arc::clone(&self.schema),
             vec![
@@ -1040,7 +1093,7 @@ impl InformationSchemataBuilder {
                 Arc::new(self.sql_path.finish()),
             ],
         )
-        .unwrap()
+        .map_err(DataFusionError::from)
     }
 }
 
@@ -1057,7 +1110,7 @@ impl PartitionStream for InformationSchemata {
             // TODO: Stream this
             futures::stream::once(async move {
                 config.make_schemata(&mut builder).await;
-                Ok(builder.finish())
+                builder.finish()
             }),
         ))
     }
@@ -1413,6 +1466,57 @@ impl PartitionStream for InformationSchemaParameters {
 mod tests {
     use super::*;
     use crate::CatalogProvider;
+    use arrow::array::Array;
+
+    #[test]
+    fn schemata_builder_emits_canonical_schema_and_rows() {
+        // Construct via `Default` so the test exercises both `new()` (via
+        // the `Default` impl) and the public column-layout contract.
+        let mut builder = InformationSchemataBuilder::default();
+        builder.add_schemata("cat", "schema_one", Some("alice"));
+        builder.add_schemata("cat", "schema_two", None);
+        let batch = builder.finish().expect("finish should not fail");
+
+        assert_eq!(batch.schema(), schemata_schema());
+        assert_eq!(batch.num_rows(), 2);
+
+        let col = |name: &str| {
+            batch
+                .column_by_name(name)
+                .unwrap_or_else(|| panic!("missing column {name}"))
+        };
+        let string_col = |name: &str| {
+            col(name)
+                .as_any()
+                .downcast_ref::<arrow::array::StringArray>()
+                .unwrap_or_else(|| panic!("{name} should be a StringArray"))
+        };
+
+        let catalog = string_col("catalog_name");
+        assert_eq!(catalog.value(0), "cat");
+        assert_eq!(catalog.value(1), "cat");
+
+        let schema = string_col("schema_name");
+        assert_eq!(schema.value(0), "schema_one");
+        assert_eq!(schema.value(1), "schema_two");
+
+        let owner = string_col("schema_owner");
+        assert_eq!(owner.value(0), "alice");
+        assert!(owner.is_null(1));
+
+        // The three character-set columns and sql_path are unconditionally
+        // null — they exist for SQL-standard column-layout compatibility.
+        for name in [
+            "default_character_set_catalog",
+            "default_character_set_schema",
+            "default_character_set_name",
+            "sql_path",
+        ] {
+            let c = string_col(name);
+            assert!(c.is_null(0), "{name} row 0 should be null");
+            assert!(c.is_null(1), "{name} row 1 should be null");
+        }
+    }
 
     #[tokio::test]
     async fn make_tables_uses_table_type() {