fix(csharp): expand SEA catalog wildcards in GetSchemas client-side (PECO-3035) [SEA]

JDBC metadata APIs (getSchemas(catalog, schemaPattern)) accept SQL LIKE-style
patterns in the catalog argument (%, _, \_), and Thrift honors those patterns
server-side. The SEA path wrapped the catalog in backticks as a literal
identifier, so SHOW SCHEMAS IN `%` returned empty rather than expanding to
all catalogs.

This change adds client-side wildcard expansion in StatementExecutionConnection:
  - null or pure '%' / '*' -> SHOW SCHEMAS IN ALL CATALOGS
  - pattern with unescaped %/_ -> SHOW CATALOGS LIKE '<pat>' then per-catalog
    SHOW SCHEMAS IN `<cat>` aggregated together
  - literal name -> unchanged (single SHOW SCHEMAS IN `<catalog>`)

Both the IGetObjectsDataProvider.GetSchemasAsync path (used by GetObjects)
and StatementExecutionStatement.GetSchemasAsync (direct GetSchemas metadata
command) now share a single ListSchemasAsync helper. Wildcard detection
honors backslash escapes (\_ stays literal).

Author: eric-wang-1990

csharp/src/StatementExecution/MetadataCommands/MetadataCommandBase.cs

@@ -99,5 +99,52 @@ protected static void AppendCatalogScope(StringBuilder sql, string? catalog)
             else
                 sql.Append(string.Format(InCatalogFormat, QuoteIdentifier(catalog)));
         }
+
+        /// <summary>
+        /// Returns true when <paramref name="pattern"/> contains a SQL LIKE wildcard
+        /// (% or _) that is NOT escaped by a preceding backslash. JDBC metadata APIs
+        /// treat catalog/schema/table arguments as LIKE patterns, but SEA SHOW commands
+        /// take literal identifiers, so callers must expand wildcards client-side.
+        /// </summary>
+        internal static bool ContainsUnescapedWildcard(string? pattern)
+        {
+            if (string.IsNullOrEmpty(pattern))
+                return false;
+
+            bool escapeNext = false;
+            for (int i = 0; i < pattern!.Length; i++)
+            {
+                char c = pattern[i];
+                if (c == '\\')
+                {
+                    // Two backslashes in a row are an escaped backslash literal, not an escape.
+                    if (i + 1 < pattern.Length && pattern[i + 1] == '\\')
+                    {
+                        i++;
+                        continue;
+                    }
+                    escapeNext = !escapeNext;
+                }
+                else if (escapeNext)
+                {
+                    escapeNext = false;
+                }
+                else if (c == '%' || c == '_')
+                {
+                    return true;
+                }
+            }
+            return false;
+        }
+
+        /// <summary>
+        /// Returns true when <paramref name="pattern"/> is a pure "match anything"
+        /// pattern: a single unescaped % (or *). These can be optimised to
+        /// SHOW SCHEMAS IN ALL CATALOGS without enumerating catalogs.
+        /// </summary>
+        internal static bool IsMatchAnything(string? pattern)
+        {
+            return pattern == "%" || pattern == "*";
+        }
     }
 }

csharp/src/StatementExecution/StatementExecutionConnection.cs

@@ -594,53 +594,10 @@ async Task<IReadOnlyList<string>> IGetObjectsDataProvider.GetCatalogsAsync(strin
 
         async Task<IReadOnlyList<(string catalog, string schema)>> IGetObjectsDataProvider.GetSchemasAsync(string? catalogPattern, string? schemaPattern, CancellationToken cancellationToken)
         {
-            // Note: catalogPattern comes from GetObjectsResultBuilder which resolves individual
-            // catalog names before calling this method. Despite the "pattern" name (from the
-            // IGetObjectsDataProvider interface), the value passed to ShowSchemasCommand is used
-            // as a literal catalog identifier (backtick-quoted), not a wildcard pattern.
-            string sql = new ShowSchemasCommand(catalogPattern, schemaPattern).Build();
-
-            List<RecordBatch> batches;
-            try
-            {
-                batches = await ExecuteMetadataSqlAsync(sql, cancellationToken).ConfigureAwait(false);
-            }
-            catch (DatabricksException ex) when (ex.IsObjectNotFoundException())
-            {
-                return System.Array.Empty<(string, string)>();
-            }
-
-            // SHOW SCHEMAS IN ALL CATALOGS returns 2 columns: databaseName, catalog
-            // SHOW SCHEMAS IN `catalog` returns 1 column: databaseName
-            bool showSchemasInAllCatalogs = catalogPattern == null;
-
-            var result = new List<(string, string)>();
-            foreach (var batch in batches)
-            {
-                StringArray? catalogArray = null;
-                StringArray? schemaArray = null;
-
-                if (showSchemasInAllCatalogs)
-                {
-                    schemaArray = batch.Column(0) as StringArray;
-                    catalogArray = batch.Column(1) as StringArray;
-                }
-                else
-                {
-                    schemaArray = batch.Column(0) as StringArray;
-                }
-
-                if (schemaArray == null) continue;
-                for (int i = 0; i < batch.Length; i++)
-                {
-                    if (schemaArray.IsNull(i)) continue;
-                    string catalog = catalogArray != null && !catalogArray.IsNull(i)
-                        ? catalogArray.GetString(i)
-                        : catalogPattern ?? "";
-                    result.Add((catalog, schemaArray.GetString(i)));
-                }
-            }
-            return result;
+            // PECO-3035: catalogPattern follows JDBC LIKE semantics (% / _ / \_ ). The SEA
+            // backend treats SHOW SCHEMAS IN `<catalog>` as a literal lookup, so we resolve
+            // wildcards client-side here (see ListSchemasAsync for details).
+            return await ListSchemasAsync(catalogPattern, schemaPattern, cancellationToken).ConfigureAwait(false);
         }
 
         async Task<IReadOnlyList<(string catalog, string schema, string table, string tableType)>> IGetObjectsDataProvider.GetTablesAsync(
@@ -785,6 +742,124 @@ internal List<RecordBatch> ExecuteMetadataSql(string sql, CancellationToken canc
             return ExecuteMetadataSqlAsync(sql, cancellationToken).GetAwaiter().GetResult();
         }
 
+        /// <summary>
+        /// Executes SHOW SCHEMAS with JDBC-style catalog pattern semantics. The SEA
+        /// backend treats <c>SHOW SCHEMAS IN `<catalog>`</c> as a literal identifier
+        /// lookup — it does not expand <c>%</c> / <c>_</c> wildcards. To match Thrift
+        /// behaviour (PECO-3035), this helper resolves wildcards client-side:
+        /// <list type="bullet">
+        ///   <item><description><c>null</c> or "%"/"*" → <c>SHOW SCHEMAS IN ALL CATALOGS</c>.</description></item>
+        ///   <item><description>A pattern containing unescaped <c>%</c> or <c>_</c> → <c>SHOW CATALOGS LIKE '&lt;pat&gt;'</c> to enumerate matching catalogs, then per-catalog <c>SHOW SCHEMAS IN `&lt;cat&gt;`</c> calls aggregated together.</description></item>
+        ///   <item><description>A literal name → single <c>SHOW SCHEMAS IN `&lt;catalog&gt;`</c> call.</description></item>
+        /// </list>
+        /// Returns a list of <c>(catalog, schema)</c> pairs in deterministic order.
+        /// </summary>
+        internal async Task<List<(string catalog, string schema)>> ListSchemasAsync(
+            string? catalogPattern, string? schemaPattern, CancellationToken cancellationToken)
+        {
+            // Fast path: null or pure "match anything" → SHOW SCHEMAS IN ALL CATALOGS.
+            if (catalogPattern == null || MetadataCommands.MetadataCommandBase.IsMatchAnything(catalogPattern))
+            {
+                return await ExecuteShowSchemasAsync(null, schemaPattern, cancellationToken).ConfigureAwait(false);
+            }
+
+            // Non-trivial wildcard: enumerate matching catalogs and iterate per-catalog.
+            if (MetadataCommands.MetadataCommandBase.ContainsUnescapedWildcard(catalogPattern))
+            {
+                string catalogsSql = new ShowCatalogsCommand(catalogPattern).Build();
+                List<RecordBatch> catalogBatches;
+                try
+                {
+                    catalogBatches = await ExecuteMetadataSqlAsync(catalogsSql, cancellationToken).ConfigureAwait(false);
+                }
+                catch (DatabricksException ex) when (ex.IsObjectNotFoundException())
+                {
+                    return new List<(string, string)>();
+                }
+
+                var matchedCatalogs = new List<string>();
+                foreach (var batch in catalogBatches)
+                {
+                    var catalogArray = batch.Column(0) as StringArray;
+                    if (catalogArray == null) continue;
+                    for (int i = 0; i < catalogArray.Length; i++)
+                    {
+                        if (!catalogArray.IsNull(i))
+                            matchedCatalogs.Add(catalogArray.GetString(i));
+                    }
+                }
+
+                var aggregated = new List<(string, string)>();
+                foreach (string cat in matchedCatalogs)
+                {
+                    try
+                    {
+                        var perCatalog = await ExecuteShowSchemasAsync(cat, schemaPattern, cancellationToken).ConfigureAwait(false);
+                        aggregated.AddRange(perCatalog);
+                    }
+                    catch (DatabricksException ex) when (ex.IsObjectNotFoundException())
+                    {
+                        // Catalog exists per SHOW CATALOGS but has no schemas / disappeared mid-query — skip.
+                    }
+                }
+                return aggregated;
+            }
+
+            // Literal catalog name.
+            return await ExecuteShowSchemasAsync(catalogPattern, schemaPattern, cancellationToken).ConfigureAwait(false);
+        }
+
+        /// <summary>
+        /// Issues a single SHOW SCHEMAS command (with the given literal catalog or
+        /// <c>IN ALL CATALOGS</c>) and decodes the result. Caller is responsible for
+        /// resolving wildcard patterns before calling this method.
+        /// </summary>
+        private async Task<List<(string catalog, string schema)>> ExecuteShowSchemasAsync(
+            string? catalog, string? schemaPattern, CancellationToken cancellationToken)
+        {
+            string sql = new ShowSchemasCommand(catalog, schemaPattern).Build();
+            List<RecordBatch> batches;
+            try
+            {
+                batches = await ExecuteMetadataSqlAsync(sql, cancellationToken).ConfigureAwait(false);
+            }
+            catch (DatabricksException ex) when (ex.IsObjectNotFoundException())
+            {
+                return new List<(string, string)>();
+            }
+
+            // SHOW SCHEMAS IN ALL CATALOGS returns 2 columns: databaseName, catalog
+            // SHOW SCHEMAS IN `catalog` returns 1 column: databaseName
+            bool showSchemasInAllCatalogs = catalog == null;
+            var result = new List<(string, string)>();
+            foreach (var batch in batches)
+            {
+                StringArray? catalogArray = null;
+                StringArray? schemaArray;
+
+                if (showSchemasInAllCatalogs)
+                {
+                    schemaArray = batch.Column(0) as StringArray;
+                    catalogArray = batch.Column(1) as StringArray;
+                }
+                else
+                {
+                    schemaArray = batch.Column(0) as StringArray;
+                }
+
+                if (schemaArray == null) continue;
+                for (int i = 0; i < batch.Length; i++)
+                {
+                    if (schemaArray.IsNull(i)) continue;
+                    string cat = catalogArray != null && !catalogArray.IsNull(i)
+                        ? catalogArray.GetString(i)
+                        : catalog ?? "";
+                    result.Add((cat, schemaArray.GetString(i)));
+                }
+            }
+            return result;
+        }
+
         /// <summary>
         /// Executes a SHOW COLUMNS command. When catalog is null, iterates over all catalogs
         /// since SHOW COLUMNS IN ALL CATALOGS is not yet supported by the backend.

csharp/src/StatementExecution/StatementExecutionStatement.cs

@@ -1076,62 +1076,25 @@ private async Task<QueryResult> GetSchemasAsync(CancellationToken cancellationTo
                     && MetadataUtilities.NormalizeSparkCatalog(_metadataCatalogName) != null)
                     return MetadataSchemaFactory.CreateEmptySchemasResult();
 
-                string sql = new ShowSchemasCommand(
+                // PECO-3035: catalog follows JDBC LIKE semantics. ListSchemasAsync expands
+                // wildcards client-side (SHOW SCHEMAS IN ALL CATALOGS or per-catalog dispatch)
+                // and returns a flat list of (catalog, schema) pairs.
+                var rows = await _connection.ListSchemasAsync(
                     catalog,
-                    EscapePatternWildcardsInName(_metadataSchemaName)).Build();
-                activity?.SetTag("sql_query", sql);
-
-                List<RecordBatch> batches;
-                try
-                {
-                    batches = await _connection.ExecuteMetadataSqlAsync(sql, cancellationToken).ConfigureAwait(false);
-                }
-                catch (DatabricksException ex) when (ex.IsObjectNotFoundException())
-                {
-                    activity?.AddEvent("statement.get_schemas.object_not_found", [
-                        new("error", ex.Message)
-                    ]);
-                    return MetadataSchemaFactory.CreateEmptySchemasResult();
-                }
-
-                // SHOW SCHEMAS IN ALL CATALOGS returns 2 columns: databaseName, catalog
-                // SHOW SCHEMAS IN `catalog` returns 1 column: databaseName
-                bool showAllCatalogs = catalog == null;
+                    EscapePatternWildcardsInName(_metadataSchemaName),
+                    cancellationToken).ConfigureAwait(false);
 
                 var tableSchemaBuilder = new StringArray.Builder();
                 var tableCatalogBuilder = new StringArray.Builder();
-                int count = 0;
-                foreach (var batch in batches)
+                foreach (var (cat, schemaName) in rows)
                 {
-                    StringArray? catalogArray = null;
-                    StringArray? schemaArray = null;
-
-                    if (showAllCatalogs)
-                    {
-                        schemaArray = batch.Column(0) as StringArray;
-                        catalogArray = batch.Column(1) as StringArray;
-                    }
-                    else
-                    {
-                        schemaArray = batch.Column(0) as StringArray;
-                    }
-
-                    if (schemaArray == null) continue;
-                    for (int i = 0; i < batch.Length; i++)
-                    {
-                        if (schemaArray.IsNull(i)) continue;
-                        tableSchemaBuilder.Append(schemaArray.GetString(i));
-                        string catalogValue = catalogArray != null && !catalogArray.IsNull(i)
-                            ? catalogArray.GetString(i)
-                            : catalog ?? "";
-                        tableCatalogBuilder.Append(catalogValue);
-                        count++;
-                    }
+                    tableSchemaBuilder.Append(schemaName);
+                    tableCatalogBuilder.Append(cat);
                 }
 
-                activity?.SetTag("result_count", count);
+                activity?.SetTag("result_count", rows.Count);
                 var schema = MetadataSchemaFactory.CreateSchemasSchema();
-                return new QueryResult(count, new HiveInfoArrowStream(schema, new IArrowArray[]
+                return new QueryResult(rows.Count, new HiveInfoArrowStream(schema, new IArrowArray[]
                 {
                     tableSchemaBuilder.Build(), tableCatalogBuilder.Build()
                 }));

csharp/test/E2E/StatementExecution/SeaMetadataE2ETests.cs

@@ -20,6 +20,7 @@
 using Apache.Arrow;
 using Apache.Arrow.Adbc;
 using Apache.Arrow.Adbc.Tests;
+using Apache.Arrow.Ipc;
 using AdbcDrivers.HiveServer2;
 using Xunit;
 using Xunit.Abstractions;
@@ -369,6 +370,75 @@ public void GetTableSchema_ThriftAndSEA_SameFieldNames()
             }
         }
 
+        // --- GetObjects: catalog wildcard pattern (PECO-3035) ---
+
+        /// <summary>
+        /// Returns the total number of (catalog, schema) pairs in a
+        /// GetObjects(depth=DbSchemas) result stream. The result schema is
+        /// [catalog_name (string), catalog_db_schemas (list&lt;struct{db_schema_name,...}&gt;)].
+        /// </summary>
+        private static async Task<int> CountSchemasInGetObjects(IArrowArrayStream stream)
+        {
+            int total = 0;
+            while (true)
+            {
+                using var batch = await stream.ReadNextRecordBatchAsync();
+                if (batch == null) break;
+                // Column 1 is the list<struct> of per-catalog schemas. Each list entry
+                // is a struct array; its Length tells us how many schemas the catalog has.
+                if (batch.Column(1) is not ListArray schemasList) continue;
+                var schemasStruct = schemasList.Values as StructArray;
+                if (schemasStruct == null) continue;
+                for (int i = 0; i < batch.Length; i++)
+                {
+                    if (schemasList.IsNull(i)) continue;
+                    int start = schemasList.ValueOffsets[i];
+                    int end = schemasList.ValueOffsets[i + 1];
+                    total += end - start;
+                }
+            }
+            return total;
+        }
+
+        [SkippableFact]
+        public async Task SEA_GetObjects_CatalogPercentWildcard_ReturnsSchemasFromAllCatalogs()
+        {
+            // PECO-3035: SEA must treat "%" in the catalog argument as a wildcard,
+            // matching Thrift / JDBC behavior. Before the fix, SEA wraps "%" in
+            // backticks and looks for a catalog literally named "%", finding no
+            // schemas → schema count = 0. With the fix, "%" expands to all catalogs
+            // and we get the full set of schemas (matching catalogPattern=null).
+            //
+            // Note: counting schemas (column 1, the list<struct>) rather than just
+            // catalogs (column 0) is what surfaces the bug — GetObjects always
+            // populates catalogs via GetCatalogsAsync (which handles "%" via
+            // SHOW CATALOGS LIKE), but GetSchemasAsync was passing "%" literally.
+            SkipIfNotConfigured();
+            using var conn = CreateSeaConnection();
+
+            using var baselineStream = conn.GetObjects(
+                depth: AdbcConnection.GetObjectsDepth.DbSchemas,
+                catalogPattern: null,
+                dbSchemaPattern: null,
+                tableNamePattern: null,
+                tableTypes: null,
+                columnNamePattern: null);
+            int baselineSchemaCount = await CountSchemasInGetObjects(baselineStream);
+
+            using var wildcardStream = conn.GetObjects(
+                depth: AdbcConnection.GetObjectsDepth.DbSchemas,
+                catalogPattern: "%",
+                dbSchemaPattern: null,
+                tableNamePattern: null,
+                tableTypes: null,
+                columnNamePattern: null);
+            int wildcardSchemaCount = await CountSchemasInGetObjects(wildcardStream);
+
+            Assert.True(baselineSchemaCount > 0,
+                "Baseline (catalogPattern=null) must return at least one schema");
+            Assert.Equal(baselineSchemaCount, wildcardSchemaCount);
+        }
+
         // --- GetTableTypes ---
 
         [SkippableFact]