BigQueryDatabaseMetaData.java

/*
 * Copyright 2024 Google LLC
 *
 * 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.
 */

package com.google.cloud.bigquery.jdbc;

import com.google.api.gax.paging.Page;
import com.google.cloud.Tuple;
import com.google.cloud.bigquery.BigQuery;
import com.google.cloud.bigquery.BigQuery.DatasetListOption;
import com.google.cloud.bigquery.BigQuery.RoutineField;
import com.google.cloud.bigquery.BigQuery.RoutineOption;
import com.google.cloud.bigquery.BigQuery.TableField;
import com.google.cloud.bigquery.BigQuery.TableOption;
import com.google.cloud.bigquery.BigQueryException;
import com.google.cloud.bigquery.ColumnReference;
import com.google.cloud.bigquery.Dataset;
import com.google.cloud.bigquery.DatasetId;
import com.google.cloud.bigquery.Field;
import com.google.cloud.bigquery.Field.Mode;
import com.google.cloud.bigquery.FieldList;
import com.google.cloud.bigquery.FieldValue;
import com.google.cloud.bigquery.FieldValueList;
import com.google.cloud.bigquery.ForeignKey;
import com.google.cloud.bigquery.PrimaryKey;
import com.google.cloud.bigquery.Routine;
import com.google.cloud.bigquery.RoutineArgument;
import com.google.cloud.bigquery.RoutineId;
import com.google.cloud.bigquery.Schema;
import com.google.cloud.bigquery.StandardSQLDataType;
import com.google.cloud.bigquery.StandardSQLField;
import com.google.cloud.bigquery.StandardSQLTableType;
import com.google.cloud.bigquery.StandardSQLTypeName;
import com.google.cloud.bigquery.Table;
import com.google.cloud.bigquery.TableConstraints;
import com.google.cloud.bigquery.TableDefinition;
import com.google.cloud.bigquery.TableId;
import com.google.cloud.bigquery.exception.BigQueryJdbcException;
import com.google.cloud.bigquery.jdbc.BigQueryTypeRegistry.ColumnTypeInfo;
import com.google.cloud.bigquery.jdbc.utils.BigQueryJdbcVersionUtility;
import com.google.common.base.Splitter;
import io.opentelemetry.context.Context;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.nio.charset.StandardCharsets;
import java.sql.Connection;
import java.sql.DatabaseMetaData;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.RowIdLifetime;
import java.sql.SQLException;
import java.sql.Statement;
import java.sql.Types;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.Comparator;
import java.util.HashSet;
import java.util.List;
import java.util.Locale;
import java.util.Set;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Callable;
import java.util.concurrent.CancellationException;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.function.Function;
import java.util.function.Supplier;
import java.util.regex.Pattern;
import javax.annotation.Nullable;

/**
 * An implementation of {@link java.sql.DatabaseMetaData}. This interface is implemented by driver
 * vendors to let users know the capabilities of a Database Management System (DBMS) in combination
 * with the driver based on JDBC™ technology ("JDBC driver") that is used with it.
 *
 * @see BigQueryStatement
 */
class BigQueryDatabaseMetaData implements DatabaseMetaData {
  static final BigQueryJdbcCustomLogger LOG =
      new BigQueryJdbcCustomLogger(BigQueryDatabaseMetaData.class.getName());
  private static final String DATABASE_PRODUCT_NAME = "Google BigQuery";
  private static final String DATABASE_PRODUCT_VERSION = "2.0";
  private static final String DRIVER_NAME = "GoogleJDBCDriverForGoogleBigQuery";
  private static final String SCHEMA_TERM = "Dataset";
  private static final String CATALOG_TERM = "Project";
  private static final String PROCEDURE_TERM = "Procedure";
  private static final int DEFAULT_PAGE_SIZE = 500;
  private static final int DEFAULT_QUEUE_CAPACITY = 5000;
  private static final String GET_EXPORTED_KEYS_SQL = "DatabaseMetaData_GetExportedKeys.sql";
  private static String exportedKeysSqlContent;
  private static final String GET_GRAPH_TABLES_REGION_SQL =
      "DatabaseMetaData_GetGraphTables_Region.sql";
  private static String graphTablesRegionSqlContent;
  private static final String GET_GRAPH_TABLES_DATASET_SQL =
      "DatabaseMetaData_GetGraphTables_Dataset.sql";
  private static String graphTablesDatasetSqlContent;
  // Declared package-private for testing.
  static final String GOOGLE_SQL_QUOTED_IDENTIFIER = "`";
  // Does not include SQL:2003 Keywords as per JDBC spec.
  // https://en.wikipedia.org/wiki/List_of_SQL_reserved_words
  static final String GOOGLE_SQL_RESERVED_KEYWORDS =
      "ASC,ASSERT_ROWS_MODIFIED,DESC,ENUM,EXCLUDE,FOLLOWING,HASH,IF,"
          + "IGNORE,LIMIT,LOOKUP,NULLS,PRECEDING,PROTO,QUALIFY,RESPECT,STRUCT,UNBOUNDED";
  static final String GOOGLE_SQL_NUMERIC_FNS =
      "ABS,ACOS,ACOSH,ASIN,ASINH,ATAN,ATAN2,ATANH,CBRT,CEIL,CEILING,COS"
          + ",COSH,COSINE_DISTANCE,COT,COTH,CSC,CSCH,DIV,EXP,EUCLIDEAN_DISTANCE,FLOOR"
          + ",GREATEST,IS_INF,LEAST,LN,LOG,LOG10,MOD,POW,RAND,RANGE_BUCKET,ROUND,"
          + ",SAFE_ADD,SAFE_DIVIDE,SAFE_MULTIPLY,SAFE_NEGATE,SAFE_SUBTRACT,SEC,SECH,"
          + "SIGN,SIN,SINH,SQRT,TAN,TANH,TRUNC";
  static final String GOOGLE_SQL_STRING_FNS =
      "ASCII,BYTE_LENGTH,CHAR_LENGTH,CHARACTER_LENGTH,CHR,CODE_POINTS_TO_BYTES,"
          + "CODE_POINTS_TO_STRING,COLLATE,CONCAT,CONTAINS_SUBSTR,EDIT_DISTANCE,ENDS_WITH,"
          + "FORMAT,FROM_BASE32,FROM_BASE64,FROM_HEX,INITCAP,INSTR,LEFT,LENGTH,LOWER,"
          + "LPAD,LTRIM,NORMALIZ,NORMALIZE_AND_CASEFOLD,OCTET_LENGTH,REGEXP_CONTAINS,"
          + "REGEXP_EXTRACT,REGEXP_EXTRACT_ALL,REGEXP_INSTR,REGEXP_REPLACE,REGEXP_SUBSTR,"
          + "REPEAT,REPLACE,REVERSE,RIGHT,RPAD,RTRIM,SAFE_CONVERT_BYTES_TO_STRING,SOUNDEX,"
          + "SPLIT,STARTS_WITH,STRPOS,SUBSTR,SUBSTRING,TO_BASE32,TO_BASE64,TO_CODE_POINTS,"
          + "TO_HEX,TRANSLATE,TRIMunicode,UNICODE,UPPER";
  static final String GOOGLE_SQL_TIME_DATE_FNS =
      "DATE,DATE_ADD,DATE_BUCKET,DATE_DIFF,DATE_FROM_UNIX_DATE,"
          + "DATE_SUB,DATE_TRUNC,DATETIME,DATETIME_ADD.,DATETIME_BUCKET,"
          + "DATETIME_DIFF,DATETIME_SUB,DATETIME_TRUNC,CURRENT_DATE,CURRENT_DATETIME,"
          + "CURRENT_TIME,CURRENT_TIMESTAMP,CURRENT_TIME,EXTRACT,FORMAT_TIME,PARSE_TIME,"
          + "TIME,TIME_ADD,TIME_DIFF,TIME_SUB,TIME_TRUNC,CURRENT_TIMESTAMP,EXTRACT,"
          + "FORMAT_TIMESTAMP,GENERATE_TIMESTAMP_ARRAY,PARSE_TIMESTAMP,TIMESTAMP,"
          + "TIMESTAMP_ADD,TIMESTAMP_DIFF,TIMESTAMP_MICROS,TIMESTAMP_MILLIS,TIMESTAMP_SECONDS,"
          + "TIMESTAMP_SUB,TIMESTAMP_TRUNC,UNIX_MICROS,UNIX_MILLIS,UNIX_SECONDS";
  static final String GOOGLE_SQL_ESCAPE = "\\";
  static final String GOOGLE_SQL_CATALOG_SEPARATOR = ".";
  static final int GOOGLE_SQL_MAX_COL_NAME_LEN = 300;
  static final int GOOGLE_SQL_MAX_COLS_PER_TABLE = 10000;
  private static final String[] DEFAULT_TABLE_TYPES = {
    "EXTERNAL", "MATERIALIZED VIEW", "SNAPSHOT", "TABLE", "VIEW"
  };
  private static final String[] TABLE_TYPES_WITH_GRAPH = {
    "EXTERNAL", "GRAPH TABLE", "MATERIALIZED VIEW", "SNAPSHOT", "TABLE", "VIEW"
  };
  static final String GRAPH_TABLE_TYPE = "GRAPH TABLE";
  private static final String DEFAULT_GRAPH_REGION = "us";
  private static final Pattern VALID_IDENTIFIER_PATTERN = Pattern.compile("^[A-Za-z0-9_\\-:.]+$");

  String URL;
  BigQueryConnection connection;
  private final BigQuery bigquery;

  BigQueryDatabaseMetaData(BigQueryConnection connection) {
    this.URL = connection.getConnectionUrl();
    this.connection = connection;
    this.bigquery = connection.getBigQuery();
  }

  @Override
  public boolean allProceduresAreCallable() {
    // Returns false because BigQuery's IAM permissions can allow a user
    // to discover a procedure's existence without having rights to execute it.
    return false;
  }

  @Override
  public boolean allTablesAreSelectable() {
    // Returns true to ensure maximum compatibility with client applications
    // that expect a positive response to discover and list all available tables.
    return true;
  }

  @Override
  public String getURL() {
    return this.URL;
  }

  @Override
  public String getUserName() {
    return null;
  }

  @Override
  public boolean isReadOnly() {
    return false;
  }

  @Override
  public boolean nullsAreSortedHigh() {
    return false;
  }

  @Override
  public boolean nullsAreSortedLow() {
    return false;
  }

  @Override
  public boolean nullsAreSortedAtStart() {
    return false;
  }

  @Override
  public boolean nullsAreSortedAtEnd() {
    return false;
  }

  @Override
  public String getDatabaseProductName() {
    return DATABASE_PRODUCT_NAME;
  }

  @Override
  public String getDatabaseProductVersion() {
    return DATABASE_PRODUCT_VERSION;
  }

  @Override
  public String getDriverName() {
    return DRIVER_NAME;
  }

  @Override
  public String getDriverVersion() {
    return BigQueryJdbcVersionUtility.getDriverVersion();
  }

  @Override
  public int getDriverMajorVersion() {
    return BigQueryJdbcVersionUtility.getDriverMajorVersion();
  }

  @Override
  public int getDriverMinorVersion() {
    return BigQueryJdbcVersionUtility.getDriverMinorVersion();
  }

  @Override
  public boolean usesLocalFiles() {
    return false;
  }

  @Override
  public boolean usesLocalFilePerTable() {
    return false;
  }

  @Override
  public boolean supportsMixedCaseIdentifiers() {
    return false;
  }

  @Override
  public boolean storesUpperCaseIdentifiers() {
    return false;
  }

  @Override
  public boolean storesLowerCaseIdentifiers() {
    return false;
  }

  @Override
  public boolean storesMixedCaseIdentifiers() {
    return false;
  }

  @Override
  public boolean supportsMixedCaseQuotedIdentifiers() {
    return false;
  }

  @Override
  public boolean storesUpperCaseQuotedIdentifiers() {
    return false;
  }

  @Override
  public boolean storesLowerCaseQuotedIdentifiers() {
    return false;
  }

  @Override
  public boolean storesMixedCaseQuotedIdentifiers() {
    return false;
  }

  @Override
  public String getIdentifierQuoteString() {
    return GOOGLE_SQL_QUOTED_IDENTIFIER;
  }

  @Override
  public String getSQLKeywords() {
    return GOOGLE_SQL_RESERVED_KEYWORDS;
  }

  @Override
  public String getNumericFunctions() {
    return GOOGLE_SQL_NUMERIC_FNS;
  }

  @Override
  public String getStringFunctions() {
    return GOOGLE_SQL_STRING_FNS;
  }

  @Override
  // GoogleSQL has UDF (user defined functions).
  // System functions like DATABASE(), USER() are not supported.
  public String getSystemFunctions() {
    return null;
  }

  @Override
  public String getTimeDateFunctions() {
    return GOOGLE_SQL_TIME_DATE_FNS;
  }

  @Override
  public String getSearchStringEscape() {
    return GOOGLE_SQL_ESCAPE;
  }

  @Override
  // No extra characters beyond a-z, A-Z, 0-9 and _
  public String getExtraNameCharacters() {
    return null;
  }

  @Override
  public boolean supportsAlterTableWithAddColumn() {
    return true;
  }

  @Override
  public boolean supportsAlterTableWithDropColumn() {
    return true;
  }

  @Override
  public boolean supportsColumnAliasing() {
    return true;
  }

  @Override
  public boolean nullPlusNonNullIsNull() {
    return true;
  }

  @Override
  public boolean supportsConvert() {
    return false;
  }

  @Override
  public boolean supportsConvert(int fromType, int toType) {
    return false;
  }

  @Override
  public boolean supportsTableCorrelationNames() {
    return true;
  }

  @Override
  public boolean supportsDifferentTableCorrelationNames() {
    return false;
  }

  @Override
  public boolean supportsExpressionsInOrderBy() {
    return true;
  }

  @Override
  public boolean supportsOrderByUnrelated() {
    return true;
  }

  @Override
  public boolean supportsGroupBy() {
    return true;
  }

  @Override
  public boolean supportsGroupByUnrelated() {
    return true;
  }

  @Override
  public boolean supportsGroupByBeyondSelect() {
    return true;
  }

  @Override
  public boolean supportsLikeEscapeClause() {
    return false;
  }

  @Override
  public boolean supportsMultipleResultSets() {
    return false;
  }

  @Override
  public boolean supportsMultipleTransactions() {
    return false;
  }

  @Override
  public boolean supportsNonNullableColumns() {
    return false;
  }

  @Override
  public boolean supportsMinimumSQLGrammar() {
    return false;
  }

  @Override
  public boolean supportsCoreSQLGrammar() {
    return false;
  }

  @Override
  public boolean supportsExtendedSQLGrammar() {
    return false;
  }

  @Override
  public boolean supportsANSI92EntryLevelSQL() {
    return false;
  }

  @Override
  public boolean supportsANSI92IntermediateSQL() {
    return false;
  }

  @Override
  public boolean supportsANSI92FullSQL() {
    return false;
  }

  @Override
  public boolean supportsIntegrityEnhancementFacility() {
    return false;
  }

  @Override
  public boolean supportsOuterJoins() {
    return false;
  }

  @Override
  public boolean supportsFullOuterJoins() {
    return false;
  }

  @Override
  public boolean supportsLimitedOuterJoins() {
    return false;
  }

  @Override
  public String getSchemaTerm() {
    return SCHEMA_TERM;
  }

  @Override
  public String getProcedureTerm() {
    return PROCEDURE_TERM;
  }

  @Override
  public String getCatalogTerm() {
    return CATALOG_TERM;
  }

  @Override
  public boolean isCatalogAtStart() {
    return true;
  }

  @Override
  public String getCatalogSeparator() {
    return GOOGLE_SQL_CATALOG_SEPARATOR;
  }

  @Override
  public boolean supportsSchemasInDataManipulation() {
    return false;
  }

  @Override
  public boolean supportsSchemasInProcedureCalls() {
    return false;
  }

  @Override
  public boolean supportsSchemasInTableDefinitions() {
    return false;
  }

  @Override
  public boolean supportsSchemasInIndexDefinitions() {
    return false;
  }

  @Override
  public boolean supportsSchemasInPrivilegeDefinitions() {
    return false;
  }

  @Override
  public boolean supportsCatalogsInDataManipulation() {
    return true;
  }

  @Override
  public boolean supportsCatalogsInProcedureCalls() {
    return true;
  }

  @Override
  public boolean supportsCatalogsInTableDefinitions() {
    return true;
  }

  @Override
  public boolean supportsCatalogsInIndexDefinitions() {
    return true;
  }

  @Override
  public boolean supportsCatalogsInPrivilegeDefinitions() {
    return true;
  }

  @Override
  public boolean supportsPositionedDelete() {
    return false;
  }

  @Override
  public boolean supportsPositionedUpdate() {
    return false;
  }

  @Override
  public boolean supportsSelectForUpdate() {
    return false;
  }

  @Override
  public boolean supportsStoredProcedures() {
    return false;
  }

  @Override
  public boolean supportsSubqueriesInComparisons() {
    return false;
  }

  @Override
  public boolean supportsSubqueriesInExists() {
    return false;
  }

  @Override
  public boolean supportsSubqueriesInIns() {
    return false;
  }

  @Override
  public boolean supportsSubqueriesInQuantifieds() {
    return false;
  }

  @Override
  public boolean supportsCorrelatedSubqueries() {
    return false;
  }

  @Override
  public boolean supportsUnion() {
    return true;
  }

  @Override
  public boolean supportsUnionAll() {
    return true;
  }

  @Override
  public boolean supportsOpenCursorsAcrossCommit() {
    return false;
  }

  @Override
  public boolean supportsOpenCursorsAcrossRollback() {
    return false;
  }

  @Override
  public boolean supportsOpenStatementsAcrossCommit() {
    return false;
  }

  @Override
  public boolean supportsOpenStatementsAcrossRollback() {
    return false;
  }

  @Override
  // No limit
  public int getMaxBinaryLiteralLength() {
    return 0;
  }

  @Override
  // No Limit
  public int getMaxCharLiteralLength() {
    return 0;
  }

  @Override
  // GoogleSQL documentation says 300.
  // https://cloud.google.com/bigquery/quotas#all_tables
  public int getMaxColumnNameLength() {
    return GOOGLE_SQL_MAX_COL_NAME_LEN;
  }

  @Override
  // No specific limits for group by.
  public int getMaxColumnsInGroupBy() {
    return 0;
  }

  @Override
  // No specific limits for index.
  public int getMaxColumnsInIndex() {
    return 0;
  }

  @Override
  // No specific limit for Order By.
  public int getMaxColumnsInOrderBy() {
    return 0;
  }

  @Override
  // All columns can be selected. No specific limits.
  public int getMaxColumnsInSelect() {
    return 0;
  }

  @Override
  public int getMaxColumnsInTable() {
    return GOOGLE_SQL_MAX_COLS_PER_TABLE;
  }

  @Override
  public int getMaxConnections() {
    // Per JDBC spec, returns 0 as there is no connection limit or is unknown.
    return 0;
  }

  @Override
  public int getMaxCursorNameLength() {
    // BigQuery does not support named cursors or positioned updates/deletes.
    return 0;
  }

  @Override
  public int getMaxIndexLength() {
    // Per the JDBC spec, 0 indicates this feature is not supported.
    return 0;
  }

  @Override
  public int getMaxSchemaNameLength() {
    // Dataset IDs can be up to 1024 characters long.
    // See: https://cloud.google.com/bigquery/docs/datasets#dataset-naming
    return 1024;
  }

  @Override
  public int getMaxProcedureNameLength() {
    // Routine IDs can be up to 256 characters long.
    // See:
    // https://cloud.google.com/bigquery/docs/reference/rest/v2/routines#RoutineReference.FIELDS.routine_id
    return 256;
  }

  @Override
  public int getMaxCatalogNameLength() {
    // Corresponds to the BigQuery Project ID, which can be a maximum of 30 characters.
    // See:
    // https://cloud.google.com/resource-manager/docs/creating-managing-projects#before_you_begin
    return 30;
  }

  @Override
  public int getMaxRowSize() {
    // Per JDBC spec, returns 0 as there is no fixed limit or is unknown.
    return 0;
  }

  @Override
  public boolean doesMaxRowSizeIncludeBlobs() {
    return false;
  }

  @Override
  public int getMaxStatementLength() {
    // Per JDBC spec, returns 0 as there is no fixed limit or is unknown.
    // See: https://cloud.google.com/bigquery/quotas#query_jobs
    return 0;
  }

  @Override
  public int getMaxStatements() {
    // Per JDBC spec, returns 0 as there is no fixed limit or is unknown.
    return 0;
  }

  @Override
  public int getMaxTableNameLength() {
    // Table IDs can be up to 1024 characters long.
    // See: https://cloud.google.com/bigquery/docs/tables#table-naming
    return 1024;
  }

  @Override
  public int getMaxTablesInSelect() {
    // BigQuery allows up to 1,000 tables to be referenced per query.
    // See: https://cloud.google.com/bigquery/quotas#query_jobs
    return 1000;
  }

  @Override
  public int getMaxUserNameLength() {
    return 0;
  }

  @Override
  public int getDefaultTransactionIsolation() {
    return Connection.TRANSACTION_SERIALIZABLE;
  }

  @Override
  public boolean supportsTransactions() {
    return true;
  }

  @Override
  public boolean supportsTransactionIsolationLevel(int level) {
    return level == Connection.TRANSACTION_SERIALIZABLE;
  }

  @Override
  public boolean supportsDataDefinitionAndDataManipulationTransactions() {
    return false;
  }

  @Override
  public boolean supportsDataManipulationTransactionsOnly() {
    return false;
  }

  @Override
  public boolean dataDefinitionCausesTransactionCommit() {
    return false;
  }

  @Override
  public boolean dataDefinitionIgnoredInTransactions() {
    return false;
  }

  @Override
  public ResultSet getProcedures(String catalog, String schemaPattern, String procedureNamePattern)
      throws SQLException {
    LOG.info(
        "getProcedures called for catalog: %s, schemaPattern: %s, procedureNamePattern: %s",
        catalog, schemaPattern, procedureNamePattern);

    final Schema resultSchema = defineGetProceduresSchema();
    if ((catalog != null && catalog.isEmpty())
        || (schemaPattern != null && schemaPattern.isEmpty())
        || (procedureNamePattern != null && procedureNamePattern.isEmpty())) {
      LOG.warning(
          "Returning empty ResultSet as one or more patterns are empty or catalog is empty.");
      return new BigQueryJsonResultSet(resultSchema);
    }

    final FieldList resultSchemaFields = resultSchema.getFields();
    List<DatasetId> targetDatasets = getTargetDatasets(catalog, schemaPattern);

    BlockingQueue<BigQueryFieldValueListWrapper> queue =
        fetchAndPopulateQueue(
            () -> {
              final List<FieldValueList> collectedResults =
                  Collections.synchronizedList(new ArrayList<>());
              processTargetRoutinesConcurrently(
                  targetDatasets,
                  procedureNamePattern,
                  true,
                  true,
                  collectedResults,
                  resultSchemaFields,
                  (routine, results, fields) -> {
                    if (!"PROCEDURE".equalsIgnoreCase(routine.getRoutineType())) {
                      return;
                    }
                    processProcedureInfo(routine, results, fields);
                  },
                  RoutineOption.fields(
                      RoutineField.ROUTINE_REFERENCE,
                      RoutineField.ARGUMENTS,
                      RoutineField.ROUTINE_TYPE));

              Comparator<FieldValueList> comparator =
                  defineGetProceduresComparator(resultSchemaFields);
              sortResults(collectedResults, comparator, "getProcedures", LOG);
              return collectedResults;
            },
            resultSchemaFields);
    return BigQueryJsonResultSet.of(resultSchema, -1, queue, null);
  }

  Schema defineGetProceduresSchema() {
    List<Field> fields = new ArrayList<>(9);
    fields.add(
        Field.newBuilder("PROCEDURE_CAT", StandardSQLTypeName.STRING)
            .setMode(Field.Mode.NULLABLE)
            .build());
    fields.add(
        Field.newBuilder("PROCEDURE_SCHEM", StandardSQLTypeName.STRING)
            .setMode(Field.Mode.NULLABLE)
            .build());
    fields.add(
        Field.newBuilder("PROCEDURE_NAME", StandardSQLTypeName.STRING)
            .setMode(Field.Mode.REQUIRED)
            .build());
    fields.add(
        Field.newBuilder("reserved1", StandardSQLTypeName.INT64)
            .setMode(Field.Mode.NULLABLE)
            .build());
    fields.add(
        Field.newBuilder("reserved2", StandardSQLTypeName.INT64)
            .setMode(Field.Mode.NULLABLE)
            .build());
    fields.add(
        Field.newBuilder("reserved3", StandardSQLTypeName.INT64)
            .setMode(Field.Mode.NULLABLE)
            .build());
    fields.add(
        Field.newBuilder("REMARKS", StandardSQLTypeName.STRING)
            .setMode(Field.Mode.NULLABLE)
            .build());
    fields.add(
        Field.newBuilder("PROCEDURE_TYPE", StandardSQLTypeName.INT64)
            .setMode(Field.Mode.REQUIRED)
            .build());
    fields.add(
        Field.newBuilder("SPECIFIC_NAME", StandardSQLTypeName.STRING)
            .setMode(Field.Mode.REQUIRED)
            .build());
    return Schema.of(fields);
  }

  void processProcedureInfo(
      Routine routine, List<FieldValueList> collectedResults, FieldList resultSchemaFields) {

    RoutineId routineId = routine.getRoutineId();
    LOG.fine("Processing procedure info for: " + routineId);

    try {
      if (!"PROCEDURE".equalsIgnoreCase(routine.getRoutineType())) {
        LOG.warning(
            "processProcedureInfo called with non-procedure type: "
                + routine.getRoutineType()
                + " for "
                + routineId);
        return;
      }

      String catalogName = routineId.getProject();
      String schemaName = routineId.getDataset();
      String procedureName = routineId.getRoutine();
      String remarks = routine.getDescription();

      List<FieldValue> values = new ArrayList<>(resultSchemaFields.size());

      values.add(createStringFieldValue(catalogName)); // 1. PROCEDURE_CAT
      values.add(createStringFieldValue(schemaName)); // 2. PROCEDURE_SCHEM
      values.add(createStringFieldValue(procedureName)); // 3. PROCEDURE_NAME
      values.add(createNullFieldValue()); // 4. reserved1
      values.add(createNullFieldValue()); // 5. reserved2
      values.add(createNullFieldValue()); // 6. reserved3
      values.add(createStringFieldValue(remarks)); // 7. REMARKS
      values.add(
          createLongFieldValue(
              (long) DatabaseMetaData.procedureResultUnknown)); // 8. PROCEDURE_TYPE
      values.add(createStringFieldValue(procedureName)); // 9. SPECIFIC_NAME

      FieldValueList rowFvl = FieldValueList.of(values, resultSchemaFields);
      collectedResults.add(rowFvl);

      LOG.fine("Processed and added procedure info row for: " + routineId);

    } catch (Exception e) {
      LOG.warning(
          "Error processing procedure info for %s: %s. Skipping this procedure.",
          routineId, e.getMessage());
    }
  }

  Comparator<FieldValueList> defineGetProceduresComparator(FieldList resultSchemaFields) {
    final int PROC_CAT_IDX = resultSchemaFields.getIndex("PROCEDURE_CAT");
    final int PROC_SCHEM_IDX = resultSchemaFields.getIndex("PROCEDURE_SCHEM");
    final int PROC_NAME_IDX = resultSchemaFields.getIndex("PROCEDURE_NAME");
    final int SPEC_NAME_IDX = resultSchemaFields.getIndex("SPECIFIC_NAME");
    return Comparator.comparing(
            (FieldValueList fvl) -> getStringValueOrNull(fvl, PROC_CAT_IDX),
            Comparator.nullsFirst(String::compareTo))
        .thenComparing(
            (FieldValueList fvl) -> getStringValueOrNull(fvl, PROC_SCHEM_IDX),
            Comparator.nullsFirst(String::compareTo))
        .thenComparing(
            (FieldValueList fvl) -> getStringValueOrNull(fvl, PROC_NAME_IDX),
            Comparator.nullsFirst(String::compareTo))
        .thenComparing(
            (FieldValueList fvl) -> getStringValueOrNull(fvl, SPEC_NAME_IDX),
            Comparator.nullsFirst(String::compareTo));
  }

  @Override
  public ResultSet getProcedureColumns(
      String catalog, String schemaPattern, String procedureNamePattern, String columnNamePattern)
      throws SQLException {
    LOG.info(
        "getProcedureColumns called for catalog: %s, schemaPattern: %s, procedureNamePattern: %s, columnNamePattern: %s",
        catalog, schemaPattern, procedureNamePattern, columnNamePattern);

    final Schema resultSchema = defineGetProcedureColumnsSchema();
    if ((catalog != null && catalog.isEmpty())
        || (schemaPattern != null && schemaPattern.isEmpty())
        || (procedureNamePattern != null && procedureNamePattern.isEmpty())
        || (columnNamePattern != null && columnNamePattern.isEmpty())) {
      LOG.warning(
          "Returning empty ResultSet as one or more patterns are empty or catalog is empty.");
      return new BigQueryJsonResultSet(resultSchema);
    }
    final FieldList resultSchemaFields = resultSchema.getFields();
    List<DatasetId> targetDatasets = getTargetDatasets(catalog, schemaPattern);
    Pattern columnNameRegex = compileSqlLikePattern(columnNamePattern);

    BlockingQueue<BigQueryFieldValueListWrapper> queue =
        fetchAndPopulateQueue(
            () -> {
              final List<FieldValueList> collectedResults =
                  Collections.synchronizedList(new ArrayList<>());
              processTargetRoutinesConcurrently(
                  targetDatasets,
                  procedureNamePattern,
                  true,
                  true,
                  collectedResults,
                  resultSchemaFields,
                  (routine, results, fields) -> {
                    if (!"PROCEDURE".equalsIgnoreCase(routine.getRoutineType())) {
                      return;
                    }
                    processProcedureArguments(routine, columnNameRegex, results, fields);
                  },
                  RoutineOption.fields(
                      RoutineField.ROUTINE_REFERENCE,
                      RoutineField.ARGUMENTS,
                      RoutineField.ROUTINE_TYPE));

              Comparator<FieldValueList> comparator =
                  defineGetProcedureColumnsComparator(resultSchemaFields);
              sortResults(collectedResults, comparator, "getProcedureColumns", LOG);
              return collectedResults;
            },
            resultSchemaFields);
    return BigQueryJsonResultSet.of(resultSchema, -1, queue, null);
  }

  Schema defineGetProcedureColumnsSchema() {
    List<Field> fields = new ArrayList<>(20);
    fields.add(
        Field.newBuilder("PROCEDURE_CAT", StandardSQLTypeName.STRING)
            .setMode(Mode.NULLABLE)
            .build());
    fields.add(
        Field.newBuilder("PROCEDURE_SCHEM", StandardSQLTypeName.STRING)
            .setMode(Mode.NULLABLE)
            .build());
    fields.add(
        Field.newBuilder("PROCEDURE_NAME", StandardSQLTypeName.STRING)
            .setMode(Mode.REQUIRED)
            .build());
    fields.add(
        Field.newBuilder("COLUMN_NAME", StandardSQLTypeName.STRING).setMode(Mode.REQUIRED).build());
    fields.add(
        Field.newBuilder("COLUMN_TYPE", StandardSQLTypeName.INT64).setMode(Mode.REQUIRED).build());
    fields.add(
        Field.newBuilder("DATA_TYPE", StandardSQLTypeName.INT64).setMode(Mode.REQUIRED).build());
    fields.add(
        Field.newBuilder("TYPE_NAME", StandardSQLTypeName.STRING).setMode(Mode.REQUIRED).build());
    fields.add(
        Field.newBuilder("PRECISION", StandardSQLTypeName.INT64).setMode(Mode.NULLABLE).build());
    fields.add(
        Field.newBuilder("LENGTH", StandardSQLTypeName.INT64).setMode(Mode.NULLABLE).build());
    fields.add(Field.newBuilder("SCALE", StandardSQLTypeName.INT64).setMode(Mode.NULLABLE).build());
    fields.add(Field.newBuilder("RADIX", StandardSQLTypeName.INT64).setMode(Mode.NULLABLE).build());
    fields.add(
        Field.newBuilder("NULLABLE", StandardSQLTypeName.INT64).setMode(Mode.REQUIRED).build());
    fields.add(
        Field.newBuilder("REMARKS", StandardSQLTypeName.STRING).setMode(Mode.NULLABLE).build());
    fields.add(
        Field.newBuilder("COLUMN_DEF", StandardSQLTypeName.STRING).setMode(Mode.NULLABLE).build());
    fields.add(
        Field.newBuilder("SQL_DATA_TYPE", StandardSQLTypeName.INT64)
            .setMode(Mode.NULLABLE)
            .build());
    fields.add(
        Field.newBuilder("SQL_DATETIME_SUB", StandardSQLTypeName.INT64)
            .setMode(Mode.NULLABLE)
            .build());
    fields.add(
        Field.newBuilder("CHAR_OCTET_LENGTH", StandardSQLTypeName.INT64)
            .setMode(Mode.NULLABLE)
            .build());
    fields.add(
        Field.newBuilder("ORDINAL_POSITION", StandardSQLTypeName.INT64)
            .setMode(Mode.REQUIRED)
            .build());
    fields.add(
        Field.newBuilder("IS_NULLABLE", StandardSQLTypeName.STRING).setMode(Mode.REQUIRED).build());
    fields.add(
        Field.newBuilder("SPECIFIC_NAME", StandardSQLTypeName.STRING)
            .setMode(Mode.REQUIRED)
            .build());
    return Schema.of(fields);
  }

  void processProcedureArguments(
      Routine routine,
      Pattern columnNameRegex,
      List<FieldValueList> collectedResults,
      FieldList resultSchemaFields) {

    RoutineId routineId = routine.getRoutineId();
    List<RoutineArgument> arguments;
    try {
      arguments = routine.getArguments();
    } catch (Exception e) {
      LOG.warning(
          "Could not retrieve arguments list for procedure %s: %s. No arguments will be"
              + " processed.",
          routineId, e.getMessage());
      return;
    }

    if (arguments == null || arguments.isEmpty()) {
      LOG.fine("Procedure " + routineId + " has no arguments.");
      return;
    }

    String catalogName = routineId.getProject();
    String schemaName = routineId.getDataset();
    String procedureName = routineId.getRoutine();
    String specificName = procedureName;

    for (int i = 0; i < arguments.size(); i++) {
      if (Thread.currentThread().isInterrupted()) {
        LOG.warning("Argument processing task interrupted for " + routineId);
        break;
      }

      int ordinalPosition = i + 1;
      RoutineArgument arg;
      String argName;

      try {
        arg = arguments.get(i);
        argName = arg.getName();
      } catch (Exception listAccessException) {
        LOG.warning(
            "Exception during arguments.get(%d) for Proc: %s. Ordinal: %d. Message: %s."
                + " Generating fallback row.",
            i, routineId, ordinalPosition, listAccessException.getMessage());
        argName = "arg_retrieval_err_" + ordinalPosition;
        arg = null;
      }

      // Filter by columnNamePattern, but not by generated fallback name
      if (columnNameRegex != null) {
        assert argName != null;
        if (!argName.startsWith("arg_")) {
          if (!columnNameRegex.matcher(argName).matches()) {
            continue;
          }
        }
      }

      List<FieldValue> values =
          createProcedureColumnRow(
              catalogName, schemaName, procedureName, specificName, arg, ordinalPosition, argName);

      FieldValueList rowFvl = FieldValueList.of(values, resultSchemaFields);
      collectedResults.add(rowFvl);
    }
  }

  List<FieldValue> createProcedureColumnRow(
      String catalog,
      String schemaName,
      String procedureName,
      String specificName,
      @Nullable RoutineArgument argument,
      int ordinalPosition,
      String columnName) {

    List<FieldValue> values = new ArrayList<>(20);
    ColumnTypeInfo typeInfo;

    if (argument == null) {
      LOG.warning(
          "Proc: %s, Arg: %s (Pos %d) - RoutineArgument object is null. Defaulting type to"
              + " VARCHAR.",
          procedureName, columnName, ordinalPosition);
      typeInfo = new ColumnTypeInfo(Types.VARCHAR, "VARCHAR", null, null, null);
    } else {
      try {
        StandardSQLDataType argumentDataType = argument.getDataType();
        if (argumentDataType == null) {
          LOG.warning(
              "Proc: %s, Arg: %s (Pos %d) - argument.getDataType() returned null. Defaulting"
                  + " type to VARCHAR.",
              procedureName, columnName, ordinalPosition);
          typeInfo = new ColumnTypeInfo(Types.VARCHAR, "VARCHAR", null, null, null);
        } else {
          typeInfo =
              determineTypeInfoFromDataType(
                  argumentDataType, procedureName, columnName, ordinalPosition);
        }
      } catch (Exception e) {
        LOG.warning(
            "Proc: %s, Arg: %s (Pos %d) - Unexpected Exception during type processing."
                + " Defaulting type to VARCHAR. Error: %s",
            procedureName, columnName, ordinalPosition, e.getMessage());
        typeInfo = new ColumnTypeInfo(Types.VARCHAR, "VARCHAR", null, null, null);
      }
    }

    String argumentModeStr = null;
    if (argument != null) {
      try {
        argumentModeStr = argument.getMode();
      } catch (Exception e) {
        LOG.warning(
            "Proc: %s, Arg: %s (Pos %d) - Could not get argument mode. Error: %s",
            procedureName, columnName, ordinalPosition, e.getMessage());
      }
    }

    values.add(createStringFieldValue(catalog)); // 1. PROCEDURE_CAT
    values.add(createStringFieldValue(schemaName)); // 2. PROCEDURE_SCHEM
    values.add(createStringFieldValue(procedureName)); // 3. PROCEDURE_NAME
    values.add(createStringFieldValue(columnName)); // 4. COLUMN_NAME
    long columnTypeJdbc = DatabaseMetaData.procedureColumnUnknown;
    if ("IN".equalsIgnoreCase(argumentModeStr)) {
      columnTypeJdbc = DatabaseMetaData.procedureColumnIn;
    } else if ("OUT".equalsIgnoreCase(argumentModeStr)) {
      columnTypeJdbc = DatabaseMetaData.procedureColumnOut;
    } else if ("INOUT".equalsIgnoreCase(argumentModeStr)) {
      columnTypeJdbc = DatabaseMetaData.procedureColumnInOut;
    }
    values.add(createLongFieldValue(columnTypeJdbc)); // 5. COLUMN_TYPE
    values.add(createLongFieldValue((long) typeInfo.jdbcType)); // 6. DATA_TYPE (java.sql.Types)
    values.add(createStringFieldValue(typeInfo.typeName)); // 7. TYPE_NAME (DB type name)
    values.add(
        createLongFieldValue(
            typeInfo.columnSize == null ? null : typeInfo.columnSize.longValue())); // 8. PRECISION
    values.add(
        createNullFieldValue()); // 9. LENGTH (length in bytes - often null for procedure params)
    values.add(
        createLongFieldValue(
            typeInfo.decimalDigits == null
                ? null
                : typeInfo.decimalDigits.longValue())); // 10. SCALE
    values.add(
        createLongFieldValue(
            typeInfo.numPrecRadix == null ? null : typeInfo.numPrecRadix.longValue())); // 11. RADIX
    values.add(createLongFieldValue((long) DatabaseMetaData.procedureNullable)); // 12. NULLABLE
    values.add(
        createNullFieldValue()); // 13. REMARKS (Can be argument.getDescription() if available and
    // needed)
    values.add(createNullFieldValue()); // 14. COLUMN_DEF (Default value - typically null)
    values.add(createNullFieldValue()); // 15. SQL_DATA_TYPE (reserved)
    values.add(createNullFieldValue()); // 16. SQL_DATETIME_SUB (reserved)
    values.add(createNullFieldValue()); // 17. CHAR_OCTET_LENGTH (null for non-char/binary)
    values.add(createLongFieldValue((long) ordinalPosition)); // 18. ORDINAL_POSITION
    values.add(createStringFieldValue("YES")); // 19. IS_NULLABLE (Default to "YES")
    values.add(createStringFieldValue(specificName)); // 20. SPECIFIC_NAME

    return values;
  }

  ColumnTypeInfo determineTypeInfoFromDataType(
      StandardSQLDataType argumentDataType,
      String procedureName,
      String columnName,
      int ordinalPosition) {

    ColumnTypeInfo defaultStringTypeInfo =
        new ColumnTypeInfo(Types.NVARCHAR, "STRING", null, null, null);
    try {
      String typeKind = argumentDataType.getTypeKind();
      if (typeKind != null && !typeKind.isEmpty()) {
        if ("ARRAY".equalsIgnoreCase(typeKind)) {
          return new ColumnTypeInfo(Types.ARRAY, "ARRAY", null, null, null);
        }
        StandardSQLTypeName determinedType = StandardSQLTypeName.valueOf(typeKind.toUpperCase());
        return getColumnTypeInfoForSqlType(determinedType);
      }
    } catch (Exception e) {
      LOG.warning(
          "Proc: %s, Arg: %s (Pos %d) - Caught an unexpected Exception during type"
              + " determination. Defaulting type to STRING. Error: %s",
          procedureName, columnName, ordinalPosition, e.getMessage());
    }
    return defaultStringTypeInfo;
  }

  Comparator<FieldValueList> defineGetProcedureColumnsComparator(FieldList resultSchemaFields) {
    final int PROC_CAT_IDX = resultSchemaFields.getIndex("PROCEDURE_CAT");
    final int PROC_SCHEM_IDX = resultSchemaFields.getIndex("PROCEDURE_SCHEM");
    final int PROC_NAME_IDX = resultSchemaFields.getIndex("PROCEDURE_NAME");
    final int SPEC_NAME_IDX = resultSchemaFields.getIndex("SPECIFIC_NAME");
    final int COL_NAME_IDX = resultSchemaFields.getIndex("COLUMN_NAME");

    if (PROC_CAT_IDX < 0
        || PROC_SCHEM_IDX < 0
        || PROC_NAME_IDX < 0
        || SPEC_NAME_IDX < 0
        || COL_NAME_IDX < 0) {
      LOG.severe(
          "Could not find required columns (PROCEDURE_CAT, SCHEM, NAME, SPECIFIC_NAME, COLUMN_NAME)"
              + " in getProcedureColumns schema for sorting. Returning null comparator.");
      return null;
    }

    return Comparator.comparing(
            (FieldValueList fvl) -> getStringValueOrNull(fvl, PROC_CAT_IDX),
            Comparator.nullsFirst(String::compareToIgnoreCase))
        .thenComparing(
            (FieldValueList fvl) -> getStringValueOrNull(fvl, PROC_SCHEM_IDX),
            Comparator.nullsFirst(String::compareToIgnoreCase))
        .thenComparing(
            (FieldValueList fvl) -> getStringValueOrNull(fvl, PROC_NAME_IDX),
            Comparator.nullsFirst(String::compareToIgnoreCase))
        .thenComparing(
            (FieldValueList fvl) -> getStringValueOrNull(fvl, SPEC_NAME_IDX),
            Comparator.nullsFirst(String::compareToIgnoreCase))
        .thenComparing(
            (FieldValueList fvl) -> getStringValueOrNull(fvl, COL_NAME_IDX),
            Comparator.nullsFirst(String::compareToIgnoreCase));
  }

  @Override
  public ResultSet getTables(
      String catalog, String schemaPattern, String tableNamePattern, String[] types)
      throws SQLException {
    return withTracing(
        "BigQueryDatabaseMetaData.getTables",
        () -> getTablesImpl(catalog, schemaPattern, tableNamePattern, types));
  }

  private ResultSet getTablesImpl(
      String catalog, String schemaPattern, String tableNamePattern, String[] types)
      throws SQLException {
    LOG.info(
        "getTables called for catalog: %s, schemaPattern: %s, tableNamePattern: %s, types: %s",
        catalog, schemaPattern, tableNamePattern, Arrays.toString(types));

    final Schema resultSchema = defineGetTablesSchema();
    if ((catalog != null && catalog.isEmpty())
        || (schemaPattern != null && schemaPattern.isEmpty())
        || (tableNamePattern != null && tableNamePattern.isEmpty())) {
      LOG.warning(
          "Returning empty ResultSet as one or more patterns are empty or catalog is empty.");
      return new BigQueryJsonResultSet(resultSchema);
    }
    final FieldList resultSchemaFields = resultSchema.getFields();
    Tuple<String, String> effectiveIdentifiers =
        determineEffectiveCatalogAndSchema(catalog, schemaPattern);
    String effectiveCatalog = effectiveIdentifiers.x();
    String effectiveSchemaPattern = effectiveIdentifiers.y();

    List<DatasetId> targetDatasets = getTargetDatasets(effectiveCatalog, effectiveSchemaPattern);

    final Set<String> requestedTypes =
        (types == null || types.length == 0) ? null : new HashSet<>(Arrays.asList(types));

    BlockingQueue<BigQueryFieldValueListWrapper> queue =
        fetchAndPopulateQueue(
            () -> {
              final List<FieldValueList> collectedResults =
                  Collections.synchronizedList(new ArrayList<>());
              boolean onlyGraphTablesRequested =
                  requestedTypes != null
                      && requestedTypes.size() == 1
                      && requestedTypes.contains(GRAPH_TABLE_TYPE);
              boolean isExactGraphTableName =
                  tableNamePattern != null
                      && !tableNamePattern.contains("%")
                      && tableNamePattern.contains(".");
              if (!isExactGraphTableName && !onlyGraphTablesRequested) {
                processTargetTablesConcurrently(
                    targetDatasets,
                    tableNamePattern,
                    true,
                    false,
                    false,
                    collectedResults,
                    resultSchemaFields,
                    (bqTable, results, fields) -> {
                      processTableInfo(bqTable, requestedTypes, results, fields);
                    },
                    TableOption.fields(
                        TableField.TABLE_REFERENCE, TableField.TYPE, TableField.DESCRIPTION));
              }

              appendGraphTableRows(
                  effectiveCatalog,
                  effectiveSchemaPattern,
                  tableNamePattern,
                  requestedTypes,
                  collectedResults,
                  resultSchemaFields);

              Comparator<FieldValueList> comparator = defineGetTablesComparator(resultSchemaFields);
              sortResults(collectedResults, comparator, "getTables", LOG);
              return collectedResults;
            },
            resultSchemaFields);

    return BigQueryJsonResultSet.of(resultSchema, -1, queue, null);
  }

  Schema defineGetTablesSchema() {
    List<Field> fields = new ArrayList<>(10);
    fields.add(
        Field.newBuilder("TABLE_CAT", StandardSQLTypeName.STRING)
            .setMode(Field.Mode.NULLABLE)
            .build());
    fields.add(
        Field.newBuilder("TABLE_SCHEM", StandardSQLTypeName.STRING)
            .setMode(Field.Mode.NULLABLE)
            .build());
    fields.add(
        Field.newBuilder("TABLE_NAME", StandardSQLTypeName.STRING)
            .setMode(Field.Mode.REQUIRED)
            .build());
    fields.add(
        Field.newBuilder("TABLE_TYPE", StandardSQLTypeName.STRING)
            .setMode(Field.Mode.REQUIRED)
            .build());
    fields.add(
        Field.newBuilder("REMARKS", StandardSQLTypeName.STRING)
            .setMode(Field.Mode.NULLABLE)
            .build());
    fields.add(
        Field.newBuilder("TYPE_CAT", StandardSQLTypeName.STRING)
            .setMode(Field.Mode.NULLABLE)
            .build());
    fields.add(
        Field.newBuilder("TYPE_SCHEM", StandardSQLTypeName.STRING)
            .setMode(Field.Mode.NULLABLE)
            .build());
    fields.add(
        Field.newBuilder("TYPE_NAME", StandardSQLTypeName.STRING)
            .setMode(Field.Mode.NULLABLE)
            .build());
    fields.add(
        Field.newBuilder("SELF_REFERENCING_COL_NAME", StandardSQLTypeName.STRING)
            .setMode(Field.Mode.NULLABLE)
            .build());
    fields.add(
        Field.newBuilder("REF_GENERATION", StandardSQLTypeName.STRING)
            .setMode(Field.Mode.NULLABLE)
            .build());
    return Schema.of(fields);
  }

  void processTableInfo(
      Table table,
      Set<String> requestedTypes,
      List<FieldValueList> collectedResults,
      FieldList resultSchemaFields) {

    TableId tableId = table.getTableId();
    LOG.fine("Processing table info for: " + tableId);

    try {
      String catalogName = tableId.getProject();
      String schemaName = tableId.getDataset();
      String tableName = tableId.getTable();
      TableDefinition definition = table.getDefinition();
      String bqTableType = definition.getType().toString();
      String remarks = table.getDescription();

      if (requestedTypes != null && !requestedTypes.contains(bqTableType)) {
        LOG.finer(
            "Skipping table %s as its type '%s' is not in the requested types %s",
            tableId, bqTableType, requestedTypes);
        return;
      }

      List<FieldValue> values = new ArrayList<>(resultSchemaFields.size());
      values.add(createStringFieldValue(catalogName)); // 1. TABLE_CAT
      values.add(createStringFieldValue(schemaName)); // 2. TABLE_SCHEM
      values.add(createStringFieldValue(tableName)); // 3. TABLE_NAME
      values.add(createStringFieldValue(bqTableType)); // 4. TABLE_TYPE
      values.add(createStringFieldValue(remarks)); // 5. REMARKS
      values.add(createNullFieldValue()); // 6. TYPE_CAT (always null)
      values.add(createNullFieldValue()); // 7. TYPE_SCHEM (always null)
      values.add(createNullFieldValue()); // 8. TYPE_NAME (always null)
      values.add(createNullFieldValue()); // 9. SELF_REFERENCING_COL_NAME (always null)
      values.add(createNullFieldValue()); // 10. REF_GENERATION (always null)

      FieldValueList rowFvl = FieldValueList.of(values, resultSchemaFields);
      collectedResults.add(rowFvl);

      LOG.fine("Processed and added table info row for: " + tableId);
    } catch (Exception e) {
      LOG.warning(
          "Error processing table info for %s: %s. Skipping this table.", tableId, e.getMessage());
    }
  }

  Comparator<FieldValueList> defineGetTablesComparator(FieldList resultSchemaFields) {
    final int TABLE_TYPE_IDX = resultSchemaFields.getIndex("TABLE_TYPE");
    final int TABLE_CAT_IDX = resultSchemaFields.getIndex("TABLE_CAT");
    final int TABLE_SCHEM_IDX = resultSchemaFields.getIndex("TABLE_SCHEM");
    final int TABLE_NAME_IDX = resultSchemaFields.getIndex("TABLE_NAME");
    return Comparator.comparing(
            (FieldValueList fvl) -> getStringValueOrNull(fvl, TABLE_TYPE_IDX),
            Comparator.nullsFirst(String::compareTo))
        .thenComparing(
            (FieldValueList fvl) -> getStringValueOrNull(fvl, TABLE_CAT_IDX),
            Comparator.nullsFirst(String::compareTo))
        .thenComparing(
            (FieldValueList fvl) -> getStringValueOrNull(fvl, TABLE_SCHEM_IDX),
            Comparator.nullsFirst(String::compareTo))
        .thenComparing(
            (FieldValueList fvl) -> getStringValueOrNull(fvl, TABLE_NAME_IDX),
            Comparator.nullsFirst(String::compareTo));
  }

  static String quoteIdentifier(String identifier) throws SQLException {
    if (identifier == null || !VALID_IDENTIFIER_PATTERN.matcher(identifier).matches()) {
      throw new BigQueryJdbcException(
          String.format("Invalid BigQuery identifier: '%s'", identifier));
    }
    return GOOGLE_SQL_QUOTED_IDENTIFIER + identifier + GOOGLE_SQL_QUOTED_IDENTIFIER;
  }

  boolean shouldSkipGraphTables(String tableNamePattern, Set<String> requestedTypes) {
    if (this.connection == null || !this.connection.isEnableGraphMetadata()) {
      return true;
    }
    if (requestedTypes != null && !requestedTypes.contains(GRAPH_TABLE_TYPE)) {
      return true;
    }
    if (tableNamePattern != null
        && !tableNamePattern.contains("%")
        && !tableNamePattern.contains(".")) {
      return true;
    }
    return false;
  }

  String resolveGraphRegion() throws SQLException {
    String location = this.connection != null ? this.connection.getLocation() : null;
    if (location == null || location.trim().isEmpty()) {
      location = DEFAULT_GRAPH_REGION;
      LOG.info(String.format("Deafulting to Graph Location %s", DEFAULT_GRAPH_REGION));
    }
    String normalized = location.trim().toLowerCase(Locale.US);
    return normalized.startsWith("region-") ? normalized : "region-" + normalized;
  }

  void appendGraphTableRows(
      String effectiveCatalog,
      String effectiveSchemaPattern,
      String tableNamePattern,
      Set<String> requestedTypes,
      List<FieldValueList> collectedResults,
      FieldList resultSchemaResults)
      throws SQLException {
    if (shouldSkipGraphTables(tableNamePattern, requestedTypes)) {
      return;
    }

    LOG.info("Appending Graph table rows");

    List<String> targetCatalogs =
        effectiveCatalog != null
            ? Collections.singletonList(effectiveCatalog)
            : getAccessibleCatalogNames();
    boolean isExactDataset =
        effectiveSchemaPattern != null && !needsListing(effectiveSchemaPattern);
    String effectiveTablePattern = (tableNamePattern == null) ? "%" : tableNamePattern;

    for (String project : targetCatalogs) {
      String formattedSql;
      if (isExactDataset) {
        formattedSql =
            replaceSqlParameters(
                getGraphTablesDatasetSqlContent(),
                quoteIdentifier(project),
                quoteIdentifier(effectiveSchemaPattern));
      } else {
        formattedSql =
            replaceSqlParameters(
                getGraphTablesRegionSqlContent(),
                quoteIdentifier(project),
                quoteIdentifier(resolveGraphRegion()));
      }

      try (PreparedStatement statement = this.connection.prepareStatement(formattedSql)) {
        if (isExactDataset) {
          statement.setString(1, effectiveTablePattern);
        } else {
          statement.setString(1, effectiveSchemaPattern == null ? "%" : effectiveSchemaPattern);
          statement.setString(2, effectiveTablePattern);
        }

        try (ResultSet rs = statement.executeQuery()) {
          while (rs.next()) {
            String tableCat = rs.getString("TABLE_CAT");
            String tableSchem = rs.getString("TABLE_SCHEM");
            String tableName = rs.getString("TABLE_NAME");
            String sourceTable = rs.getString("SOURCE_TABLE");
            String remarks = sourceTable != null ? "Source: " + sourceTable : null;

            List<FieldValue> values = new ArrayList<>(resultSchemaResults.size());
            values.add(createStringFieldValue(tableCat));
            values.add(createStringFieldValue(tableSchem));
            values.add(createStringFieldValue(tableName));
            values.add(createStringFieldValue(GRAPH_TABLE_TYPE));
            values.add(createStringFieldValue(remarks));
            values.add(createNullFieldValue());
            values.add(createNullFieldValue());
            values.add(createNullFieldValue());
            values.add(createNullFieldValue());
            values.add(createNullFieldValue());
            collectedResults.add(FieldValueList.of(values, resultSchemaResults));
          }
        }
      } catch (SQLException e) {
        if (e.getMessage() != null && e.getMessage().contains("Not found")) {
          LOG.info(
              "Dataset or project not found while querying PROPERTY_GRAPHS in project '%s'."
                  + " Skipping.",
              project);
        } else {
          throw e;
        }
      }
    }
  }

  @Override
  public ResultSet getSchemas() throws SQLException {
    LOG.info("getSchemas() called");

    return getSchemas(null, null);
  }

  @Override
  public ResultSet getCatalogs() throws SQLException {
    return withTracing("BigQueryDatabaseMetaData.getCatalogs", () -> getCatalogsImpl());
  }

  private ResultSet getCatalogsImpl() throws SQLException {
    LOG.info("getCatalogs() called");

    final List<String> accessibleCatalogs = getAccessibleCatalogNames();
    final Schema catalogsSchema = defineGetCatalogsSchema();
    final FieldList schemaFields = catalogsSchema.getFields();
    final List<FieldValueList> catalogRows =
        prepareGetCatalogsRows(schemaFields, accessibleCatalogs);

    BlockingQueue<BigQueryFieldValueListWrapper> queue =
        fetchAndPopulateQueue(() -> catalogRows, schemaFields);

    return BigQueryJsonResultSet.of(catalogsSchema, catalogRows.size(), queue, null);
  }

  Schema defineGetCatalogsSchema() {
    return Schema.of(
        Field.newBuilder("TABLE_CAT", StandardSQLTypeName.STRING).setMode(Mode.REQUIRED).build());
  }

  List<FieldValueList> prepareGetCatalogsRows(
      FieldList schemaFields, List<String> accessibleCatalogs) {
    List<FieldValueList> catalogRows = new ArrayList<>();
    for (String catalogName : accessibleCatalogs) {
      FieldValue fieldValue = FieldValue.of(FieldValue.Attribute.PRIMITIVE, catalogName);
      catalogRows.add(FieldValueList.of(Collections.singletonList(fieldValue), schemaFields));
    }
    return catalogRows;
  }

  @Override
  public ResultSet getTableTypes() throws SQLException {
    LOG.info("getTableTypes() called");

    final Schema tableTypesSchema = defineGetTableTypesSchema();
    final List<FieldValueList> tableTypeRows =
        prepareGetTableTypesRows(tableTypesSchema, this.connection.isEnableGraphMetadata());

    BlockingQueue<BigQueryFieldValueListWrapper> queue =
        fetchAndPopulateQueue(() -> tableTypeRows, tableTypesSchema.getFields());

    return BigQueryJsonResultSet.of(tableTypesSchema, tableTypeRows.size(), queue, null);
  }

  static Schema defineGetTableTypesSchema() {
    return Schema.of(
        Field.newBuilder("TABLE_TYPE", StandardSQLTypeName.STRING)
            .setMode(Field.Mode.REQUIRED)
            .build());
  }

  static List<FieldValueList> prepareGetTableTypesRows(Schema schema) {
    return prepareGetTableTypesRows(schema, false);
  }

  static List<FieldValueList> prepareGetTableTypesRows(Schema schema, boolean includeGraphTables) {
    final String[] tableTypes = includeGraphTables ? TABLE_TYPES_WITH_GRAPH : DEFAULT_TABLE_TYPES;
    FieldList schemaFields = schema.getFields();
    List<FieldValueList> rows = new ArrayList<>(tableTypes.length);

    for (String typeName : tableTypes) {
      FieldValue fieldValue = FieldValue.of(FieldValue.Attribute.PRIMITIVE, typeName);
      rows.add(FieldValueList.of(Collections.singletonList(fieldValue), schemaFields));
    }
    return rows;
  }

  @Override
  public ResultSet getColumns(
      String catalog, String schemaPattern, String tableNamePattern, String columnNamePattern)
      throws SQLException {
    return withTracing(
        "BigQueryDatabaseMetaData.getColumns",
        () -> getColumnsImpl(catalog, schemaPattern, tableNamePattern, columnNamePattern));
  }

  private ResultSet getColumnsImpl(
      String catalog, String schemaPattern, String tableNamePattern, String columnNamePattern)
      throws SQLException {
    LOG.info(
        "getColumns called for catalog: %s, schemaPattern: %s, tableNamePattern: %s, columnNamePattern: %s",
        catalog, schemaPattern, tableNamePattern, columnNamePattern);

    final Schema resultSchema = defineGetColumnsSchema();
    if ((catalog != null && catalog.isEmpty())
        || (schemaPattern != null && schemaPattern.isEmpty())
        || (tableNamePattern != null && tableNamePattern.isEmpty())
        || (columnNamePattern != null && columnNamePattern.isEmpty())) {
      LOG.warning(
          "Returning empty ResultSet as one or more patterns are empty or catalog is empty.");
      return new BigQueryJsonResultSet(resultSchema);
    }
    final FieldList resultSchemaFields = resultSchema.getFields();
    Tuple<String, String> effectiveIdentifiers =
        determineEffectiveCatalogAndSchema(catalog, schemaPattern);
    String effectiveCatalog = effectiveIdentifiers.x();
    String effectiveSchemaPattern = effectiveIdentifiers.y();

    List<DatasetId> targetDatasets = getTargetDatasets(effectiveCatalog, effectiveSchemaPattern);

    Pattern columnNameRegex = compileSqlLikePattern(columnNamePattern);

    BlockingQueue<BigQueryFieldValueListWrapper> queue =
        fetchAndPopulateQueue(
            () -> {
              final List<FieldValueList> collectedResults =
                  Collections.synchronizedList(new ArrayList<>());
              processTargetTablesConcurrently(
                  targetDatasets,
                  tableNamePattern,
                  true,
                  false,
                  true,
                  collectedResults,
                  resultSchemaFields,
                  (bqTable, results, fields) -> {
                    processTableColumns(bqTable, columnNameRegex, results, fields);
                  },
                  TableOption.fields(
                      TableField.TABLE_REFERENCE, TableField.TYPE, TableField.SCHEMA));

              Comparator<FieldValueList> comparator =
                  defineGetColumnsComparator(resultSchemaFields);
              sortResults(collectedResults, comparator, "getColumns", LOG);
              return collectedResults;
            },
            resultSchemaFields);

    return BigQueryJsonResultSet.of(resultSchema, -1, queue, null);
  }

  private void processTableColumns(
      Table table,
      Pattern columnNameRegex,
      List<FieldValueList> collectedResults,
      FieldList resultSchemaFields) {
    TableId tableId = table.getTableId();
    LOG.fine("Processing columns for table: " + tableId);
    TableDefinition definition = table.getDefinition();
    Schema tableSchema = (definition != null) ? definition.getSchema() : null;

    try {
      if (tableSchema == null) {
        LOG.fine(
            "Schema not included in table object for "
                + tableId
                + ", fetching full table details...");
        Table fullTable =
            bigquery.getTable(
                tableId,
                TableOption.fields(TableField.TABLE_REFERENCE, TableField.TYPE, TableField.SCHEMA));
        if (fullTable != null) {
          definition = fullTable.getDefinition();
          tableSchema = (definition != null) ? definition.getSchema() : null;
        } else {
          LOG.warning(
              "Table " + tableId + " not found when fetching full details for columns. Skipping.");
          return;
        }
      }

      if (tableSchema == null
          || tableSchema.getFields() == null
          || tableSchema.getFields().isEmpty()) {
        LOG.warning(
            "Schema not found or fields are null for table %s (Type: %s). Skipping columns.",
            tableId, definition.getType());
        return;
      }

      FieldList fields = tableSchema.getFields();
      String catalogName = tableId.getProject();
      String schemaName = tableId.getDataset();
      String tableName = tableId.getTable();

      for (int i = 0; i < fields.size(); i++) {
        if (Thread.currentThread().isInterrupted()) {
          LOG.warning("Task for table " + tableId + " interrupted during column iteration.");
          break;
        }
        Field field = fields.get(i);
        String currentColumnName = field.getName();
        if (columnNameRegex != null && !columnNameRegex.matcher(currentColumnName).matches())
          continue;
        List<FieldValue> values = createColumnRow(catalogName, schemaName, tableName, field, i + 1);
        FieldValueList rowFvl = FieldValueList.of(values, resultSchemaFields);
        collectedResults.add(rowFvl);
      }
      LOG.fine("Finished processing columns for table: " + tableId);
    } catch (BigQueryException e) {
      LOG.warning(
          "BigQueryException processing table %s: %s (Code: %d)",
          tableId, e.getMessage(), e.getCode());
    } catch (Exception e) {
      LOG.severe("Unexpected error processing table %s: %s", tableId, e.getMessage());
    }
  }

  private Schema defineGetColumnsSchema() {
    List<Field> fields = new ArrayList<>(24);
    fields.add(
        Field.newBuilder("TABLE_CAT", StandardSQLTypeName.STRING)
            .setMode(Field.Mode.NULLABLE)
            .build()); // 1
    fields.add(
        Field.newBuilder("TABLE_SCHEM", StandardSQLTypeName.STRING)
            .setMode(Field.Mode.NULLABLE)
            .build()); // 2
    fields.add(
        Field.newBuilder("TABLE_NAME", StandardSQLTypeName.STRING)
            .setMode(Field.Mode.REQUIRED)
            .build()); // 3
    fields.add(
        Field.newBuilder("COLUMN_NAME", StandardSQLTypeName.STRING)
            .setMode(Field.Mode.REQUIRED)
            .build()); // 4
    fields.add(
        Field.newBuilder("DATA_TYPE", StandardSQLTypeName.INT64)
            .setMode(Field.Mode.REQUIRED)
            .build()); // 5
    fields.add(
        Field.newBuilder("TYPE_NAME", StandardSQLTypeName.STRING)
            .setMode(Field.Mode.REQUIRED)
            .build()); // 6
    fields.add(
        Field.newBuilder("COLUMN_SIZE", StandardSQLTypeName.INT64)
            .setMode(Field.Mode.NULLABLE)
            .build()); // 7
    fields.add(
        Field.newBuilder("BUFFER_LENGTH", StandardSQLTypeName.INT64)
            .setMode(Field.Mode.NULLABLE)
            .build()); // 8
    fields.add(
        Field.newBuilder("DECIMAL_DIGITS", StandardSQLTypeName.INT64)
            .setMode(Field.Mode.NULLABLE)
            .build()); // 9
    fields.add(
        Field.newBuilder("NUM_PREC_RADIX", StandardSQLTypeName.INT64)
            .setMode(Field.Mode.NULLABLE)
            .build()); // 10
    fields.add(
        Field.newBuilder("NULLABLE", StandardSQLTypeName.INT64)
            .setMode(Field.Mode.REQUIRED)
            .build()); // 11
    fields.add(
        Field.newBuilder("REMARKS", StandardSQLTypeName.STRING)
            .setMode(Field.Mode.NULLABLE)
            .build()); // 12
    fields.add(
        Field.newBuilder("COLUMN_DEF", StandardSQLTypeName.STRING)
            .setMode(Field.Mode.NULLABLE)
            .build()); // 13
    fields.add(
        Field.newBuilder("SQL_DATA_TYPE", StandardSQLTypeName.INT64)
            .setMode(Field.Mode.NULLABLE)
            .build()); // 14
    fields.add(
        Field.newBuilder("SQL_DATETIME_SUB", StandardSQLTypeName.INT64)
            .setMode(Field.Mode.NULLABLE)
            .build()); // 15
    fields.add(
        Field.newBuilder("CHAR_OCTET_LENGTH", StandardSQLTypeName.INT64)
            .setMode(Field.Mode.NULLABLE)
            .build()); // 16
    fields.add(
        Field.newBuilder("ORDINAL_POSITION", StandardSQLTypeName.INT64)
            .setMode(Field.Mode.REQUIRED)
            .build()); // 17
    fields.add(
        Field.newBuilder("IS_NULLABLE", StandardSQLTypeName.STRING)
            .setMode(Field.Mode.REQUIRED)
            .build()); // 18
    fields.add(
        Field.newBuilder("SCOPE_CATALOG", StandardSQLTypeName.STRING)
            .setMode(Field.Mode.NULLABLE)
            .build()); // 19
    fields.add(
        Field.newBuilder("SCOPE_SCHEMA", StandardSQLTypeName.STRING)
            .setMode(Field.Mode.NULLABLE)
            .build()); // 20
    fields.add(
        Field.newBuilder("SCOPE_TABLE", StandardSQLTypeName.STRING)
            .setMode(Field.Mode.NULLABLE)
            .build()); // 21
    fields.add(
        Field.newBuilder("SOURCE_DATA_TYPE", StandardSQLTypeName.INT64)
            .setMode(Field.Mode.NULLABLE)
            .build()); // 22
    fields.add(
        Field.newBuilder("IS_AUTOINCREMENT", StandardSQLTypeName.STRING)
            .setMode(Field.Mode.REQUIRED)
            .build()); // 23
    fields.add(
        Field.newBuilder("IS_GENERATEDCOLUMN", StandardSQLTypeName.STRING)
            .setMode(Field.Mode.REQUIRED)
            .build()); // 24
    return Schema.of(fields);
  }

  List<FieldValue> createColumnRow(
      String catalog, String schemaName, String tableName, Field field, int ordinalPosition) {
    List<FieldValue> values = new ArrayList<>(24);
    Field.Mode mode = (field.getMode() == null) ? Field.Mode.NULLABLE : field.getMode();
    ColumnTypeInfo typeInfo = mapBigQueryTypeToJdbc(field);

    values.add(createStringFieldValue(catalog)); // 1. TABLE_CAT
    values.add(createStringFieldValue(schemaName)); // 2. TABLE_SCHEM
    values.add(createStringFieldValue(tableName)); // 3. TABLE_NAME
    values.add(createStringFieldValue(field.getName())); // 4. COLUMN_NAME
    values.add(createLongFieldValue((long) typeInfo.jdbcType)); // 5. DATA_TYPE
    values.add(createStringFieldValue(typeInfo.typeName)); // 6. TYPE_NAME
    values.add(
        createLongFieldValue(
            typeInfo.columnSize == null
                ? null
                : typeInfo.columnSize.longValue())); // 7. COLUMN_SIZE
    values.add(createNullFieldValue()); // 8. BUFFER_LENGTH
    values.add(
        createLongFieldValue(
            typeInfo.decimalDigits == null
                ? null
                : typeInfo.decimalDigits.longValue())); // 9. DECIMAL_DIGITS
    values.add(
        createLongFieldValue(
            typeInfo.numPrecRadix == null
                ? null
                : typeInfo.numPrecRadix.longValue())); // 10. NUM_PREC_RADIX
    int nullable =
        (mode == Field.Mode.REQUIRED)
            ? DatabaseMetaData.columnNoNulls
            : DatabaseMetaData.columnNullable;
    values.add(createLongFieldValue((long) nullable)); // 11. NULLABLE
    values.add(createStringFieldValue(field.getDescription())); // 12. REMARKS
    values.add(createNullFieldValue()); // 13. COLUMN_DEF
    values.add(createNullFieldValue()); // 14. SQL_DATA_TYPE
    values.add(createNullFieldValue()); // 15. SQL_DATETIME_SUB
    values.add(createNullFieldValue()); // 16. CHAR_OCTET_LENGTH
    values.add(createLongFieldValue((long) ordinalPosition)); // 17. ORDINAL_POSITION
    String isNullable = "";
    switch (mode) {
      case REQUIRED:
        isNullable = "NO";
        break;
      case NULLABLE:
      case REPEATED:
        isNullable = "YES";
        break;
    }
    values.add(createStringFieldValue(isNullable)); // 18. IS_NULLABLE
    values.add(createNullFieldValue()); // 19. SCOPE_CATALOG
    values.add(createNullFieldValue()); // 20. SCOPE_SCHEMA
    values.add(createNullFieldValue()); // 21. SCOPE_TABLE
    values.add(createNullFieldValue()); // 22. SOURCE_DATA_TYPE
    values.add(createStringFieldValue("NO")); // 23. IS_AUTOINCREMENT
    values.add(createStringFieldValue("NO")); // 24. IS_GENERATEDCOLUMN

    return values;
  }

  ColumnTypeInfo mapBigQueryTypeToJdbc(Field field) {
    Mode mode = (field.getMode() == null) ? Mode.NULLABLE : field.getMode();
    if (mode == Mode.REPEATED) {
      return new ColumnTypeInfo(Types.ARRAY, "ARRAY", null, null, null);
    }

    if (this.connection.isEnableTimestampPicos()
        && BigQueryTemporalUtility.isPicosecondTimestamp(field)) {
      return new ColumnTypeInfo(
          Types.VARCHAR, BigQueryTemporalUtility.TIMESTAMP_PICOSECONDS_TYPE_NAME, 32, 12, null);
    }

    StandardSQLTypeName bqType = null;
    if (field.getType() != null && field.getType().getStandardType() != null) {
      bqType = field.getType().getStandardType();
    }
    return getColumnTypeInfoForSqlType(bqType);
  }

  private Comparator<FieldValueList> defineGetColumnsComparator(FieldList resultSchemaFields) {
    final int TABLE_CAT_IDX = resultSchemaFields.getIndex("TABLE_CAT");
    final int TABLE_SCHEM_IDX = resultSchemaFields.getIndex("TABLE_SCHEM");
    final int TABLE_NAME_IDX = resultSchemaFields.getIndex("TABLE_NAME");
    final int ORDINAL_POS_IDX = resultSchemaFields.getIndex("ORDINAL_POSITION");
    return Comparator.comparing(
            (FieldValueList fvl) -> getStringValueOrNull(fvl, TABLE_CAT_IDX),
            Comparator.nullsFirst(String::compareTo))
        .thenComparing(
            (FieldValueList fvl) -> getStringValueOrNull(fvl, TABLE_SCHEM_IDX),
            Comparator.nullsFirst(String::compareTo))
        .thenComparing(
            (FieldValueList fvl) -> getStringValueOrNull(fvl, TABLE_NAME_IDX),
            Comparator.nullsFirst(String::compareTo))
        .thenComparing(
            (FieldValueList fvl) -> getLongValueOrNull(fvl, ORDINAL_POS_IDX),
            Comparator.nullsFirst(Long::compareTo));
  }

  @Override
  public ResultSet getColumnPrivileges(
      String catalog, String schema, String table, String columnNamePattern) {
    LOG.info(
        "getColumnPrivileges called for catalog: %s, schema: %s, table: %s, columnNamePattern:"
            + " %s. BigQuery IAM model differs from SQL privileges; returning empty ResultSet.",
        catalog, schema, table, columnNamePattern);

    return new BigQueryJsonResultSet(defineGetColumnPrivilegesSchema());
  }

  Schema defineGetColumnPrivilegesSchema() {
    List<Field> fields = defineBasePrivilegeFields();

    Field columnNameField =
        Field.newBuilder("COLUMN_NAME", StandardSQLTypeName.STRING)
            .setMode(Field.Mode.REQUIRED)
            .build();
    fields.add(3, columnNameField);

    return Schema.of(fields);
  }

  @Override
  public ResultSet getTablePrivileges(
      String catalog, String schemaPattern, String tableNamePattern) {
    LOG.info(
        "getTablePrivileges called for catalog: %s, schemaPattern: %s, tableNamePattern: %s. "
            + "BigQuery IAM model differs from SQL privileges; returning empty ResultSet.",
        catalog, schemaPattern, tableNamePattern);

    return new BigQueryJsonResultSet(defineGetTablePrivilegesSchema());
  }

  Schema defineGetTablePrivilegesSchema() {
    List<Field> fields = defineBasePrivilegeFields();
    return Schema.of(fields);
  }

  @Override
  public ResultSet getBestRowIdentifier(
      String catalog, String schema, String table, int scope, boolean nullable) {
    LOG.info(
        "getBestRowIdentifier called for catalog: %s, schema: %s, table: %s, scope: %d,"
            + " nullable: %s. BigQuery does not support best row identifiers; returning empty"
            + " ResultSet.",
        catalog, schema, table, scope, nullable);

    return new BigQueryJsonResultSet(defineGetBestRowIdentifierSchema());
  }

  Schema defineGetBestRowIdentifierSchema() {
    List<Field> fields = new ArrayList<>(8);
    fields.add(
        Field.newBuilder("SCOPE", StandardSQLTypeName.INT64).setMode(Field.Mode.REQUIRED).build());
    fields.add(
        Field.newBuilder("COLUMN_NAME", StandardSQLTypeName.STRING)
            .setMode(Field.Mode.REQUIRED)
            .build());
    fields.add(
        Field.newBuilder("DATA_TYPE", StandardSQLTypeName.INT64)
            .setMode(Field.Mode.REQUIRED)
            .build());
    fields.add(
        Field.newBuilder("TYPE_NAME", StandardSQLTypeName.STRING)
            .setMode(Field.Mode.REQUIRED)
            .build());
    fields.add(
        Field.newBuilder("COLUMN_SIZE", StandardSQLTypeName.INT64)
            .setMode(Field.Mode.NULLABLE)
            .build());
    fields.add(
        Field.newBuilder("BUFFER_LENGTH", StandardSQLTypeName.INT64)
            .setMode(Field.Mode.NULLABLE)
            .build());
    fields.add(
        Field.newBuilder("DECIMAL_DIGITS", StandardSQLTypeName.INT64)
            .setMode(Field.Mode.NULLABLE)
            .build());
    fields.add(
        Field.newBuilder("PSEUDO_COLUMN", StandardSQLTypeName.INT64)
            .setMode(Field.Mode.REQUIRED)
            .build());
    return Schema.of(fields);
  }

  @Override
  public ResultSet getVersionColumns(String catalog, String schema, String table) {
    LOG.info(
        "getVersionColumns called for catalog: %s, schema: %s, table: %s. "
            + "Automatic version columns not supported by BigQuery; returning empty ResultSet.",
        catalog, schema, table);

    return new BigQueryJsonResultSet(defineGetVersionColumnsSchema());
  }

  Schema defineGetVersionColumnsSchema() {
    List<Field> fields = new ArrayList<>(8);
    fields.add(
        Field.newBuilder("SCOPE", StandardSQLTypeName.INT64).setMode(Field.Mode.NULLABLE).build());
    fields.add(
        Field.newBuilder("COLUMN_NAME", StandardSQLTypeName.STRING)
            .setMode(Field.Mode.REQUIRED)
            .build());
    fields.add(
        Field.newBuilder("DATA_TYPE", StandardSQLTypeName.INT64)
            .setMode(Field.Mode.REQUIRED)
            .build());
    fields.add(
        Field.newBuilder("TYPE_NAME", StandardSQLTypeName.STRING)
            .setMode(Field.Mode.REQUIRED)
            .build());
    fields.add(
        Field.newBuilder("COLUMN_SIZE", StandardSQLTypeName.INT64)
            .setMode(Field.Mode.NULLABLE)
            .build());
    fields.add(
        Field.newBuilder("BUFFER_LENGTH", StandardSQLTypeName.INT64)
            .setMode(Field.Mode.REQUIRED)
            .build());
    fields.add(
        Field.newBuilder("DECIMAL_DIGITS", StandardSQLTypeName.INT64)
            .setMode(Field.Mode.NULLABLE)
            .build());
    fields.add(
        Field.newBuilder("PSEUDO_COLUMN", StandardSQLTypeName.INT64)
            .setMode(Field.Mode.REQUIRED)
            .build());
    return Schema.of(fields);
  }

  @Override
  public ResultSet getPrimaryKeys(String catalog, String schema, String table) throws SQLException {
    LOG.info(
        "getPrimaryKeys called for catalog: %s, schema: %s, table: %s", catalog, schema, table);
    final Schema resultSchema = defineGetPrimaryKeysSchema();
    if ((catalog != null && catalog.isEmpty())
        || (schema != null && schema.isEmpty())
        || table == null
        || table.isEmpty()) {
      LOG.warning(
          "Returning empty ResultSet as one or more patterns are empty or catalog is empty.");
      return new BigQueryJsonResultSet(resultSchema);
    }

    final FieldList resultSchemaFields = resultSchema.getFields();
    List<DatasetId> targetDatasets = getTargetDatasets(catalog, schema, false);

    BlockingQueue<BigQueryFieldValueListWrapper> queue =
        fetchAndPopulateQueue(
            () -> {
              final List<FieldValueList> collectedResults =
                  Collections.synchronizedList(new ArrayList<>());
              processTargetTablesConcurrently(
                  targetDatasets,
                  table,
                  false,
                  true,
                  true,
                  collectedResults,
                  resultSchemaFields,
                  (bqTable, results, fields) -> {
                    TableConstraints constraints = bqTable.getTableConstraints();
                    processPrimaryKey(constraints, bqTable.getTableId(), results, fields);
                  });

              Comparator<FieldValueList> comparator =
                  defineGetPrimaryKeysComparator(resultSchemaFields);
              sortResults(collectedResults, comparator, "getPrimaryKeys", LOG);
              return collectedResults;
            },
            resultSchemaFields);

    return BigQueryJsonResultSet.of(resultSchema, -1, queue, null);
  }

  private Schema defineGetPrimaryKeysSchema() {
    List<Field> fields = new ArrayList<>(6);
    fields.add(
        Field.newBuilder("TABLE_CAT", StandardSQLTypeName.STRING)
            .setMode(Field.Mode.NULLABLE)
            .build());
    fields.add(
        Field.newBuilder("TABLE_SCHEM", StandardSQLTypeName.STRING)
            .setMode(Field.Mode.NULLABLE)
            .build());
    fields.add(
        Field.newBuilder("TABLE_NAME", StandardSQLTypeName.STRING)
            .setMode(Field.Mode.REQUIRED)
            .build());
    fields.add(
        Field.newBuilder("COLUMN_NAME", StandardSQLTypeName.STRING)
            .setMode(Field.Mode.REQUIRED)
            .build());
    fields.add(
        Field.newBuilder("KEY_SEQ", StandardSQLTypeName.INT64)
            .setMode(Field.Mode.REQUIRED)
            .build());
    fields.add(
        Field.newBuilder("PK_NAME", StandardSQLTypeName.STRING)
            .setMode(Field.Mode.NULLABLE)
            .build());
    return Schema.of(fields);
  }

  private void processPrimaryKey(
      TableConstraints constraints,
      TableId tableId,
      List<FieldValueList> collectedResults,
      FieldList resultSchemaFields) {
    if (constraints == null || constraints.getPrimaryKey() == null) {
      return;
    }
    PrimaryKey pk = constraints.getPrimaryKey();
    List<String> pkColumns = pk.getColumns();
    for (int i = 0; i < pkColumns.size(); i++) {
      List<FieldValue> row = new ArrayList<>(6);
      row.add(createStringFieldValue(tableId.getProject())); // 1. TABLE_CAT
      row.add(createStringFieldValue(tableId.getDataset())); // 2. TABLE_SCHEM
      row.add(createStringFieldValue(tableId.getTable())); // 3. TABLE_NAME
      row.add(createStringFieldValue(pkColumns.get(i))); // 4. COLUMN_NAME
      row.add(createLongFieldValue((long) (i + 1))); // 5. KEY_SEQ
      row.add(createNullFieldValue()); // 6. PK_NAME
      collectedResults.add(FieldValueList.of(row, resultSchemaFields));
    }
  }

  private Comparator<FieldValueList> defineGetPrimaryKeysComparator(FieldList resultSchemaFields) {
    final int COLUMN_NAME_IDX = resultSchemaFields.getIndex("COLUMN_NAME");
    return Comparator.comparing(
        (FieldValueList fvl) -> getStringValueOrNull(fvl, COLUMN_NAME_IDX),
        Comparator.nullsFirst(String::compareTo));
  }

  @Override
  public ResultSet getImportedKeys(String catalog, String schema, String table)
      throws SQLException {
    LOG.info(
        "getImportedKeys called for catalog: %s, schema: %s, table: %s", catalog, schema, table);
    final Schema resultSchema = defineForeignKeyResultSetSchema();
    if ((catalog != null && catalog.isEmpty())
        || (schema != null && schema.isEmpty())
        || table == null
        || table.isEmpty()) {
      LOG.warning(
          "Returning empty ResultSet as one or more patterns are empty or catalog is empty.");
      return new BigQueryJsonResultSet(resultSchema);
    }

    final FieldList resultSchemaFields = resultSchema.getFields();
    List<DatasetId> targetDatasets = getTargetDatasets(catalog, schema, false);

    BlockingQueue<BigQueryFieldValueListWrapper> queue =
        fetchAndPopulateQueue(
            () -> {
              final List<FieldValueList> collectedResults =
                  Collections.synchronizedList(new ArrayList<>());
              processTargetTablesConcurrently(
                  targetDatasets,
                  table,
                  false,
                  true,
                  true,
                  collectedResults,
                  resultSchemaFields,
                  (bqTable, results, fields) -> {
                    TableConstraints constraints = bqTable.getTableConstraints();
                    if (constraints == null || constraints.getForeignKeys() == null) {
                      return;
                    }
                    for (ForeignKey fk : constraints.getForeignKeys()) {
                      TableId pkTableId = fk.getReferencedTable();
                      processForeignKey(fk, pkTableId, bqTable.getTableId(), results, fields);
                    }
                  });

              Comparator<FieldValueList> comparator =
                  definePkTableSortComparator(resultSchemaFields);
              sortResults(collectedResults, comparator, "getImportedKeys", LOG);
              return collectedResults;
            },
            resultSchemaFields);
    return BigQueryJsonResultSet.of(resultSchema, -1, queue, null);
  }

  @Override
  public ResultSet getExportedKeys(String catalog, String schema, String table)
      throws SQLException {
    LOG.info(
        "getExportedKeys called for catalog: %s, schema: %s, table: %s", catalog, schema, table);
    final Schema resultSchema = defineForeignKeyResultSetSchema();
    if ((catalog != null && catalog.isEmpty())
        || (schema != null && schema.isEmpty())
        || table == null
        || table.isEmpty()) {
      LOG.warning(
          "Returning empty ResultSet as one or more patterns are empty or catalog is empty.");
      return new BigQueryJsonResultSet(resultSchema);
    }

    final FieldList resultSchemaFields = resultSchema.getFields();

    // Early return for PCNT catalog schemas (containing '.') as they do not support table
    // constraints.
    if (schema != null && schema.contains(".")) {
      return new BigQueryJsonResultSet(resultSchema);
    }

    // Fallback Path: If catalog or schema is null, fall back to REST API metadata scan.
    if (catalog == null || schema == null) {
      List<DatasetId> targetDatasets = getTargetDatasets(catalog, schema, false);
      BlockingQueue<BigQueryFieldValueListWrapper> queue =
          fetchAndPopulateQueue(
              () -> {
                final List<FieldValueList> collectedResults =
                    Collections.synchronizedList(new ArrayList<>());
                processTargetTablesConcurrently(
                    targetDatasets,
                    null,
                    false,
                    true,
                    true,
                    collectedResults,
                    resultSchemaFields,
                    (bqTable, results, fields) -> {
                      TableConstraints constraints = bqTable.getTableConstraints();
                      if (constraints == null || constraints.getForeignKeys() == null) {
                        return;
                      }
                      for (ForeignKey fk : constraints.getForeignKeys()) {
                        TableId pkTableId = fk.getReferencedTable();
                        if (pkTableId == null
                            || !equalsOrNullMatchesAll(catalog, pkTableId.getProject())
                            || !equalsOrNullMatchesAll(schema, pkTableId.getDataset())
                            || !table.equals(pkTableId.getTable())) {
                          continue;
                        }
                        processForeignKey(fk, pkTableId, bqTable.getTableId(), results, fields);
                      }
                    });

                Comparator<FieldValueList> comparator =
                    defineFkTableSortComparator(resultSchemaFields);
                sortResults(collectedResults, comparator, "getExportedKeys", LOG);
                return collectedResults;
              },
              resultSchemaFields);
      return BigQueryJsonResultSet.of(resultSchema, -1, queue, null);
    }

    String sql = getExportedKeysSqlContent();
    String formattedSql = replaceSqlParameters(sql, catalog, schema, table);
    PreparedStatement stmt = this.connection.prepareStatement(formattedSql);
    if (stmt == null) {
      throw new BigQueryJdbcException("Failed to prepare statement for getExportedKeys");
    }
    try {
      stmt.closeOnCompletion();
      return stmt.executeQuery();
    } catch (SQLException e) {
      closeStatementIgnoreException(stmt);
      throw new BigQueryJdbcException("Error executing getExportedKeys", e);
    }
  }

  @Override
  public ResultSet getCrossReference(
      String parentCatalog,
      String parentSchema,
      String parentTable,
      String foreignCatalog,
      String foreignSchema,
      String foreignTable)
      throws SQLException {
    LOG.info(
        "getCrossReference called for parentCatalog: %s, parentSchema: %s, parentTable: %s, foreignCatalog: %s, foreignSchema: %s, foreignTable: %s",
        parentCatalog, parentSchema, parentTable, foreignCatalog, foreignSchema, foreignTable);
    final Schema resultSchema = defineForeignKeyResultSetSchema();
    if ((parentCatalog != null && parentCatalog.isEmpty())
        || (parentSchema != null && parentSchema.isEmpty())
        || parentTable == null
        || parentTable.isEmpty()
        || (foreignCatalog != null && foreignCatalog.isEmpty())
        || (foreignSchema != null && foreignSchema.isEmpty())
        || foreignTable == null
        || foreignTable.isEmpty()) {
      LOG.warning(
          "Returning empty ResultSet as one or more patterns are empty or catalog is empty.");
      return new BigQueryJsonResultSet(resultSchema);
    }

    final FieldList resultSchemaFields = resultSchema.getFields();
    List<DatasetId> targetDatasets = getTargetDatasets(foreignCatalog, foreignSchema, false);

    BlockingQueue<BigQueryFieldValueListWrapper> queue =
        fetchAndPopulateQueue(
            () -> {
              final List<FieldValueList> collectedResults =
                  Collections.synchronizedList(new ArrayList<>());
              processTargetTablesConcurrently(
                  targetDatasets,
                  foreignTable,
                  false,
                  true,
                  true,
                  collectedResults,
                  resultSchemaFields,
                  (bqTable, results, fields) -> {
                    TableConstraints constraints = bqTable.getTableConstraints();
                    if (constraints == null || constraints.getForeignKeys() == null) {
                      return;
                    }
                    for (ForeignKey fk : constraints.getForeignKeys()) {
                      TableId pkTableId = fk.getReferencedTable();
                      if (!equalsOrNullMatchesAll(parentCatalog, pkTableId.getProject())
                          || !equalsOrNullMatchesAll(parentSchema, pkTableId.getDataset())
                          || !parentTable.equals(pkTableId.getTable())) {
                        continue;
                      }
                      processForeignKey(fk, pkTableId, bqTable.getTableId(), results, fields);
                    }
                  });

              Comparator<FieldValueList> comparator =
                  defineFkTableSortComparator(resultSchemaFields);
              sortResults(collectedResults, comparator, "getCrossReference", LOG);
              return collectedResults;
            },
            resultSchemaFields);
    return BigQueryJsonResultSet.of(resultSchema, -1, queue, null);
  }

  @Override
  public ResultSet getTypeInfo() throws SQLException {
    LOG.info("getTypeInfo() called");

    final Schema typeInfoSchema = defineGetTypeInfoSchema();
    final FieldList schemaFields = typeInfoSchema.getFields();
    final List<FieldValueList> typeInfoRows = prepareGetTypeInfoRows(schemaFields);

    final Comparator<FieldValueList> comparator = defineGetTypeInfoComparator(schemaFields);
    sortResults(typeInfoRows, comparator, "getTypeInfo", LOG);
    BlockingQueue<BigQueryFieldValueListWrapper> queue =
        fetchAndPopulateQueue(() -> typeInfoRows, schemaFields);
    return BigQueryJsonResultSet.of(typeInfoSchema, typeInfoRows.size(), queue, null);
  }

  Schema defineGetTypeInfoSchema() {
    List<Field> fields = new ArrayList<>(18);
    fields.add(
        Field.newBuilder("TYPE_NAME", StandardSQLTypeName.STRING)
            .setMode(Mode.REQUIRED)
            .build()); // 1
    fields.add(
        Field.newBuilder("DATA_TYPE", StandardSQLTypeName.INT64)
            .setMode(Mode.REQUIRED)
            .build()); // 2
    fields.add(
        Field.newBuilder("PRECISION", StandardSQLTypeName.INT64)
            .setMode(Mode.NULLABLE)
            .build()); // 3
    fields.add(
        Field.newBuilder("LITERAL_PREFIX", StandardSQLTypeName.STRING)
            .setMode(Mode.NULLABLE)
            .build()); // 4
    fields.add(
        Field.newBuilder("LITERAL_SUFFIX", StandardSQLTypeName.STRING)
            .setMode(Mode.NULLABLE)
            .build()); // 5
    fields.add(
        Field.newBuilder("CREATE_PARAMS", StandardSQLTypeName.STRING)
            .setMode(Mode.NULLABLE)
            .build()); // 6
    fields.add(
        Field.newBuilder("NULLABLE", StandardSQLTypeName.INT64)
            .setMode(Mode.REQUIRED)
            .build()); // 7
    fields.add(
        Field.newBuilder("CASE_SENSITIVE", StandardSQLTypeName.BOOL)
            .setMode(Mode.REQUIRED)
            .build()); // 8
    fields.add(
        Field.newBuilder("SEARCHABLE", StandardSQLTypeName.INT64)
            .setMode(Mode.REQUIRED)
            .build()); // 9
    fields.add(
        Field.newBuilder("UNSIGNED_ATTRIBUTE", StandardSQLTypeName.BOOL)
            .setMode(Mode.NULLABLE)
            .build()); // 10
    fields.add(
        Field.newBuilder("FIXED_PREC_SCALE", StandardSQLTypeName.BOOL)
            .setMode(Mode.REQUIRED)
            .build()); // 11
    fields.add(
        Field.newBuilder("AUTO_INCREMENT", StandardSQLTypeName.BOOL)
            .setMode(Mode.REQUIRED)
            .build()); // 12
    fields.add(
        Field.newBuilder("LOCAL_TYPE_NAME", StandardSQLTypeName.STRING)
            .setMode(Mode.NULLABLE)
            .build()); // 13
    fields.add(
        Field.newBuilder("MINIMUM_SCALE", StandardSQLTypeName.INT64)
            .setMode(Mode.NULLABLE)
            .build()); // 14
    fields.add(
        Field.newBuilder("MAXIMUM_SCALE", StandardSQLTypeName.INT64)
            .setMode(Mode.NULLABLE)
            .build()); // 15
    fields.add(
        Field.newBuilder("SQL_DATA_TYPE", StandardSQLTypeName.INT64)
            .setMode(Mode.NULLABLE)
            .build()); // 16
    fields.add(
        Field.newBuilder("SQL_DATETIME_SUB", StandardSQLTypeName.INT64)
            .setMode(Mode.NULLABLE)
            .build()); // 17
    fields.add(
        Field.newBuilder("NUM_PREC_RADIX", StandardSQLTypeName.INT64)
            .setMode(Mode.NULLABLE)
            .build()); // 18
    return Schema.of(fields);
  }

  List<FieldValueList> prepareGetTypeInfoRows(FieldList schemaFields) {
    List<FieldValueList> rows = new ArrayList<>();

    Function<TypeInfoRowData, FieldValueList> createRow =
        (data) -> {
          List<FieldValue> values = new ArrayList<>(18);
          values.add(createStringFieldValue(data.typeName)); // 1. TYPE_NAME
          values.add(createLongFieldValue((long) data.jdbcType)); // 2. DATA_TYPE
          values.add(createLongFieldValue(data.precision)); // 3. PRECISION
          values.add(createStringFieldValue(data.literalPrefix)); // 4. LITERAL_PREFIX
          values.add(createStringFieldValue(data.literalSuffix)); // 5. LITERAL_SUFFIX
          values.add(createStringFieldValue(data.createParams)); // 6. CREATE_PARAMS
          values.add(createLongFieldValue((long) data.nullable)); // 7. NULLABLE
          values.add(createBooleanFieldValue(data.caseSensitive)); // 8. CASE_SENSITIVE
          values.add(createLongFieldValue((long) data.searchable)); // 9. SEARCHABLE
          values.add(createBooleanFieldValue(data.unsignedAttribute)); // 10. UNSIGNED_ATTRIBUTE
          values.add(createBooleanFieldValue(data.fixedPrecScale)); // 11. FIXED_PREC_SCALE
          values.add(createBooleanFieldValue(data.autoIncrement)); // 12. AUTO_INCREMENT
          values.add(createStringFieldValue(data.localTypeName)); // 13. LOCAL_TYPE_NAME
          values.add(createLongFieldValue(data.minimumScale)); // 14. MINIMUM_SCALE
          values.add(createLongFieldValue(data.maximumScale)); // 15. MAXIMUM_SCALE
          values.add(createNullFieldValue()); // 16. SQL_DATA_TYPE
          values.add(createNullFieldValue()); // 17. SQL_DATETIME_SUB
          values.add(createLongFieldValue(data.numPrecRadix)); // 18. NUM_PREC_RADIX
          return FieldValueList.of(values, schemaFields);
        };

    rows.add(
        createRow.apply(
            new TypeInfoRowData(
                "INT64",
                Types.BIGINT,
                19L,
                null,
                null,
                null,
                DatabaseMetaData.typeNullable,
                false,
                DatabaseMetaData.typeSearchable,
                false,
                false,
                false,
                "INT64",
                0L,
                0L,
                10L)));
    rows.add(
        createRow.apply(
            new TypeInfoRowData(
                "BOOL",
                Types.BOOLEAN,
                1L,
                null,
                null,
                null,
                DatabaseMetaData.typeNullable,
                false,
                DatabaseMetaData.typePredBasic,
                false,
                false,
                false,
                "BOOL",
                0L,
                0L,
                null)));
    rows.add(
        createRow.apply(
            new TypeInfoRowData(
                "FLOAT64",
                Types.DOUBLE,
                15L,
                null,
                null,
                null,
                DatabaseMetaData.typeNullable,
                false,
                DatabaseMetaData.typeSearchable,
                false,
                false,
                false,
                "FLOAT64",
                null,
                null,
                2L)));
    rows.add(
        createRow.apply(
            new TypeInfoRowData(
                "NUMERIC",
                Types.NUMERIC,
                38L,
                null,
                null,
                "PRECISION,SCALE",
                DatabaseMetaData.typeNullable,
                false,
                DatabaseMetaData.typeSearchable,
                false,
                true,
                false,
                "NUMERIC",
                9L,
                9L,
                10L)));
    rows.add(
        createRow.apply(
            new TypeInfoRowData(
                "BIGNUMERIC",
                Types.NUMERIC,
                77L,
                null,
                null,
                "PRECISION,SCALE",
                DatabaseMetaData.typeNullable,
                false,
                DatabaseMetaData.typeSearchable,
                false,
                true,
                false,
                "BIGNUMERIC",
                38L,
                38L,
                10L)));
    rows.add(
        createRow.apply(
            new TypeInfoRowData(
                "STRING",
                Types.NVARCHAR,
                null,
                "'",
                "'",
                "LENGTH",
                DatabaseMetaData.typeNullable,
                true,
                DatabaseMetaData.typeSearchable,
                false,
                false,
                false,
                "STRING",
                null,
                null,
                null)));
    rows.add(
        createRow.apply(
            new TypeInfoRowData(
                "TIMESTAMP",
                Types.TIMESTAMP,
                29L,
                "'",
                "'",
                null,
                DatabaseMetaData.typeNullable,
                false,
                DatabaseMetaData.typeSearchable,
                false,
                false,
                false,
                "TIMESTAMP",
                null,
                null,
                null)));
    rows.add(
        createRow.apply(
            new TypeInfoRowData(
                "DATE",
                Types.DATE,
                10L,
                "'",
                "'",
                null,
                DatabaseMetaData.typeNullable,
                false,
                DatabaseMetaData.typeSearchable,
                false,
                false,
                false,
                "DATE",
                null,
                null,
                null)));
    rows.add(
        createRow.apply(
            new TypeInfoRowData(
                "TIME",
                Types.TIME,
                15L,
                "'",
                "'",
                null,
                DatabaseMetaData.typeNullable,
                false,
                DatabaseMetaData.typeSearchable,
                false,
                false,
                false,
                "TIME",
                null,
                null,
                null)));
    rows.add(
        createRow.apply(
            new TypeInfoRowData(
                "DATETIME",
                Types.TIMESTAMP,
                29L,
                "'",
                "'",
                null,
                DatabaseMetaData.typeNullable,
                false,
                DatabaseMetaData.typeSearchable,
                false,
                false,
                false,
                "DATETIME",
                null,
                null,
                null)));
    rows.add(
        createRow.apply(
            new TypeInfoRowData(
                "GEOGRAPHY",
                Types.OTHER,
                null,
                "'",
                "'",
                null,
                DatabaseMetaData.typeNullable,
                false,
                DatabaseMetaData.typeSearchable,
                false,
                false,
                false,
                "GEOGRAPHY",
                null,
                null,
                null)));
    rows.add(
        createRow.apply(
            new TypeInfoRowData(
                "JSON",
                Types.OTHER,
                null,
                "'",
                "'",
                null,
                DatabaseMetaData.typeNullable,
                false,
                DatabaseMetaData.typeSearchable,
                false,
                false,
                false,
                "JSON",
                null,
                null,
                null)));
    rows.add(
        createRow.apply(
            new TypeInfoRowData(
                "INTERVAL",
                Types.OTHER,
                null,
                "'",
                "'",
                null,
                DatabaseMetaData.typeNullable,
                false,
                DatabaseMetaData.typeSearchable,
                false,
                false,
                false,
                "INTERVAL",
                null,
                null,
                null)));
    rows.add(
        createRow.apply(
            new TypeInfoRowData(
                "BYTES",
                Types.VARBINARY,
                null,
                "0x",
                null,
                "LENGTH",
                DatabaseMetaData.typeNullable,
                false,
                DatabaseMetaData.typeSearchable,
                false,
                false,
                false,
                "BYTES",
                null,
                null,
                null)));
    rows.add(
        createRow.apply(
            new TypeInfoRowData(
                "STRUCT",
                Types.STRUCT,
                null,
                null,
                null,
                null,
                DatabaseMetaData.typeNullable,
                false,
                DatabaseMetaData.typePredNone,
                false,
                false,
                false,
                "STRUCT",
                null,
                null,
                null)));
    rows.add(
        createRow.apply(
            new TypeInfoRowData(
                "ARRAY",
                Types.ARRAY,
                null,
                null,
                null,
                null,
                DatabaseMetaData.typeNullable,
                false,
                DatabaseMetaData.typePredNone,
                false,
                false,
                false,
                "ARRAY",
                null,
                null,
                null)));
    rows.add(
        createRow.apply(
            new TypeInfoRowData(
                "RANGE",
                Types.OTHER,
                null,
                null,
                null,
                null,
                DatabaseMetaData.typeNullable,
                false,
                DatabaseMetaData.typeSearchable,
                false,
                false,
                false,
                "RANGE",
                null,
                null,
                null)));

    return rows;
  }

  Comparator<FieldValueList> defineGetTypeInfoComparator(FieldList schemaFields) {
    final int DATA_TYPE_IDX = schemaFields.getIndex("DATA_TYPE");
    if (DATA_TYPE_IDX < 0) {
      LOG.severe(
          "Could not find DATA_TYPE column in getTypeInfo schema for sorting. Returning null"
              + " comparator.");
      return null;
    }
    Comparator<FieldValueList> comparator =
        Comparator.comparing(
            (FieldValueList fvl) -> getLongValueOrNull(fvl, DATA_TYPE_IDX),
            Comparator.nullsFirst(Long::compareTo));
    return comparator;
  }

  @Override
  public ResultSet getIndexInfo(
      String catalog, String schema, String table, boolean unique, boolean approximate) {
    LOG.info(
        "getIndexInfo called for catalog: %s, schema: %s, table: %s, unique: %s, approximate:"
            + " %s. Traditional indexes not supported by BigQuery; returning empty ResultSet.",
        catalog, schema, table, unique, approximate);

    return new BigQueryJsonResultSet(defineGetIndexInfoSchema());
  }

  Schema defineGetIndexInfoSchema() {
    List<Field> fields = new ArrayList<>(13);
    fields.add(
        Field.newBuilder("TABLE_CAT", StandardSQLTypeName.STRING)
            .setMode(Field.Mode.NULLABLE)
            .build());
    fields.add(
        Field.newBuilder("TABLE_SCHEM", StandardSQLTypeName.STRING)
            .setMode(Field.Mode.NULLABLE)
            .build());
    fields.add(
        Field.newBuilder("TABLE_NAME", StandardSQLTypeName.STRING)
            .setMode(Field.Mode.REQUIRED)
            .build());
    fields.add(
        Field.newBuilder("NON_UNIQUE", StandardSQLTypeName.BOOL)
            .setMode(Field.Mode.REQUIRED)
            .build());
    fields.add(
        Field.newBuilder("INDEX_QUALIFIER", StandardSQLTypeName.STRING)
            .setMode(Field.Mode.NULLABLE)
            .build());
    fields.add(
        Field.newBuilder("INDEX_NAME", StandardSQLTypeName.STRING)
            .setMode(Field.Mode.NULLABLE)
            .build());
    fields.add(
        Field.newBuilder("TYPE", StandardSQLTypeName.INT64).setMode(Field.Mode.REQUIRED).build());
    fields.add(
        Field.newBuilder("ORDINAL_POSITION", StandardSQLTypeName.INT64)
            .setMode(Field.Mode.REQUIRED)
            .build());
    fields.add(
        Field.newBuilder("COLUMN_NAME", StandardSQLTypeName.STRING)
            .setMode(Field.Mode.NULLABLE)
            .build());
    fields.add(
        Field.newBuilder("ASC_OR_DESC", StandardSQLTypeName.STRING)
            .setMode(Field.Mode.NULLABLE)
            .build());
    fields.add(
        Field.newBuilder("CARDINALITY", StandardSQLTypeName.INT64)
            .setMode(Field.Mode.NULLABLE)
            .build());
    fields.add(
        Field.newBuilder("PAGES", StandardSQLTypeName.INT64).setMode(Field.Mode.NULLABLE).build());
    fields.add(
        Field.newBuilder("FILTER_CONDITION", StandardSQLTypeName.STRING)
            .setMode(Field.Mode.NULLABLE)
            .build());
    return Schema.of(fields);
  }

  @Override
  public boolean supportsResultSetType(int type) {
    // BigQuery primarily supports forward-only result sets.
    return type == ResultSet.TYPE_FORWARD_ONLY;
  }

  @Override
  public boolean supportsResultSetConcurrency(int type, int concurrency) {
    // BigQuery primarily supports forward-only, read-only result sets.
    return type == ResultSet.TYPE_FORWARD_ONLY && concurrency == ResultSet.CONCUR_READ_ONLY;
  }

  @Override
  public boolean ownUpdatesAreVisible(int type) {
    return false;
  }

  @Override
  public boolean ownDeletesAreVisible(int type) {
    return false;
  }

  @Override
  public boolean ownInsertsAreVisible(int type) {
    return false;
  }

  @Override
  public boolean othersUpdatesAreVisible(int type) {
    return false;
  }

  @Override
  public boolean othersDeletesAreVisible(int type) {
    return false;
  }

  @Override
  public boolean othersInsertsAreVisible(int type) {
    return false;
  }

  @Override
  public boolean updatesAreDetected(int type) {
    return false;
  }

  @Override
  public boolean deletesAreDetected(int type) {
    return false;
  }

  @Override
  public boolean insertsAreDetected(int type) {
    return false;
  }

  @Override
  public boolean supportsBatchUpdates() {
    return true;
  }

  @Override
  public ResultSet getUDTs(
      String catalog, String schemaPattern, String typeNamePattern, int[] types) {
    LOG.info(
        "getUDTs called for catalog: %s, schemaPattern: %s, typeNamePattern: %s, types: %s. "
            + "Feature not supported by BigQuery; returning empty ResultSet.",
        catalog, schemaPattern, typeNamePattern, (types == null ? "null" : Arrays.toString(types)));

    return new BigQueryJsonResultSet(defineGetUDTsSchema());
  }

  Schema defineGetUDTsSchema() {
    List<Field> fields = new ArrayList<>(7);
    fields.add(
        Field.newBuilder("TYPE_CAT", StandardSQLTypeName.STRING)
            .setMode(Field.Mode.NULLABLE)
            .build());
    fields.add(
        Field.newBuilder("TYPE_SCHEM", StandardSQLTypeName.STRING)
            .setMode(Field.Mode.NULLABLE)
            .build());
    fields.add(
        Field.newBuilder("TYPE_NAME", StandardSQLTypeName.STRING)
            .setMode(Field.Mode.REQUIRED)
            .build());
    fields.add(
        Field.newBuilder("CLASS_NAME", StandardSQLTypeName.STRING)
            .setMode(Field.Mode.REQUIRED)
            .build());
    fields.add(
        Field.newBuilder("DATA_TYPE", StandardSQLTypeName.INT64)
            .setMode(Field.Mode.REQUIRED)
            .build());
    fields.add(
        Field.newBuilder("REMARKS", StandardSQLTypeName.STRING)
            .setMode(Field.Mode.NULLABLE)
            .build());
    fields.add(
        Field.newBuilder("BASE_TYPE", StandardSQLTypeName.INT64)
            .setMode(Field.Mode.NULLABLE)
            .build());
    return Schema.of(fields);
  }

  @Override
  public Connection getConnection() {
    return connection;
  }

  @Override
  public boolean supportsSavepoints() {
    return false;
  }

  @Override
  public boolean supportsNamedParameters() {
    return false;
  }

  @Override
  public boolean supportsMultipleOpenResults() {
    return false;
  }

  @Override
  public boolean supportsGetGeneratedKeys() {
    return false;
  }

  @Override
  public ResultSet getSuperTables(String catalog, String schemaPattern, String tableNamePattern) {
    LOG.info(
        "getSuperTables called for catalog: %s, schemaPattern: %s, tableNamePattern: %s. "
            + "BigQuery does not support super tables; returning empty ResultSet.",
        catalog, schemaPattern, tableNamePattern);

    return new BigQueryJsonResultSet(defineGetSuperTablesSchema());
  }

  Schema defineGetSuperTablesSchema() {
    List<Field> fields = new ArrayList<>(4);
    fields.add(
        Field.newBuilder("TABLE_CAT", StandardSQLTypeName.STRING)
            .setMode(Field.Mode.NULLABLE)
            .build()); // 1. TABLE_CAT
    fields.add(
        Field.newBuilder("TABLE_SCHEM", StandardSQLTypeName.STRING)
            .setMode(Field.Mode.NULLABLE)
            .build()); // 2. TABLE_SCHEM
    fields.add(
        Field.newBuilder("TABLE_NAME", StandardSQLTypeName.STRING)
            .setMode(Field.Mode.REQUIRED)
            .build()); // 3. TABLE_NAME
    fields.add(
        Field.newBuilder("SUPERTABLE_NAME", StandardSQLTypeName.STRING)
            .setMode(Field.Mode.REQUIRED)
            .build()); // 4. SUPERTABLE_NAME
    return Schema.of(fields);
  }

  @Override
  public ResultSet getSuperTypes(String catalog, String schemaPattern, String typeNamePattern) {
    LOG.info(
        "getSuperTypes called for catalog: %s, schemaPattern: %s, typeNamePattern: %s. BigQuery"
            + " does not support user-defined type hierarchies; returning empty ResultSet.",
        catalog, schemaPattern, typeNamePattern);

    return new BigQueryJsonResultSet(defineGetSuperTypesSchema());
  }

  Schema defineGetSuperTypesSchema() {
    List<Field> fields = new ArrayList<>(6);
    fields.add(
        Field.newBuilder("TYPE_CAT", StandardSQLTypeName.STRING)
            .setMode(Field.Mode.NULLABLE)
            .build()); // 1. TYPE_CAT
    fields.add(
        Field.newBuilder("TYPE_SCHEM", StandardSQLTypeName.STRING)
            .setMode(Field.Mode.NULLABLE)
            .build()); // 2. TYPE_SCHEM
    fields.add(
        Field.newBuilder("TYPE_NAME", StandardSQLTypeName.STRING)
            .setMode(Field.Mode.REQUIRED)
            .build()); // 3. TYPE_NAME
    fields.add(
        Field.newBuilder("SUPERTYPE_CAT", StandardSQLTypeName.STRING)
            .setMode(Field.Mode.NULLABLE)
            .build()); // 4. SUPERTYPE_CAT
    fields.add(
        Field.newBuilder("SUPERTYPE_SCHEM", StandardSQLTypeName.STRING)
            .setMode(Field.Mode.NULLABLE)
            .build()); // 5. SUPERTYPE_SCHEM
    fields.add(
        Field.newBuilder("SUPERTYPE_NAME", StandardSQLTypeName.STRING)
            .setMode(Field.Mode.REQUIRED)
            .build()); // 6. SUPERTYPE_NAME
    return Schema.of(fields);
  }

  @Override
  public ResultSet getAttributes(
      String catalog, String schemaPattern, String typeNamePattern, String attributeNamePattern) {
    LOG.info(
        "getAttributes called for catalog: %s, schemaPattern: %s, typeNamePattern: %s,"
            + " attributeNamePattern: %s. Feature not supported by BigQuery; returning empty"
            + " ResultSet.",
        catalog, schemaPattern, typeNamePattern, attributeNamePattern);

    return new BigQueryJsonResultSet(defineGetAttributesSchema());
  }

  Schema defineGetAttributesSchema() {
    List<Field> fields = new ArrayList<>(21);
    fields.add(
        Field.newBuilder("TYPE_CAT", StandardSQLTypeName.STRING)
            .setMode(Field.Mode.NULLABLE)
            .build());
    fields.add(
        Field.newBuilder("TYPE_SCHEM", StandardSQLTypeName.STRING)
            .setMode(Field.Mode.NULLABLE)
            .build());
    fields.add(
        Field.newBuilder("TYPE_NAME", StandardSQLTypeName.STRING)
            .setMode(Field.Mode.REQUIRED)
            .build());
    fields.add(
        Field.newBuilder("ATTR_NAME", StandardSQLTypeName.STRING)
            .setMode(Field.Mode.REQUIRED)
            .build());
    fields.add(
        Field.newBuilder("DATA_TYPE", StandardSQLTypeName.INT64)
            .setMode(Field.Mode.REQUIRED)
            .build());
    fields.add(
        Field.newBuilder("ATTR_TYPE_NAME", StandardSQLTypeName.STRING)
            .setMode(Field.Mode.REQUIRED)
            .build());
    fields.add(
        Field.newBuilder("ATTR_SIZE", StandardSQLTypeName.INT64)
            .setMode(Field.Mode.NULLABLE)
            .build());
    fields.add(
        Field.newBuilder("DECIMAL_DIGITS", StandardSQLTypeName.INT64)
            .setMode(Field.Mode.NULLABLE)
            .build());
    fields.add(
        Field.newBuilder("NUM_PREC_RADIX", StandardSQLTypeName.INT64)
            .setMode(Field.Mode.NULLABLE)
            .build());
    fields.add(
        Field.newBuilder("NULLABLE", StandardSQLTypeName.INT64)
            .setMode(Field.Mode.REQUIRED)
            .build());
    fields.add(
        Field.newBuilder("REMARKS", StandardSQLTypeName.STRING)
            .setMode(Field.Mode.NULLABLE)
            .build());
    fields.add(
        Field.newBuilder("ATTR_DEF", StandardSQLTypeName.STRING)
            .setMode(Field.Mode.NULLABLE)
            .build());
    fields.add(
        Field.newBuilder("SQL_DATA_TYPE", StandardSQLTypeName.INT64)
            .setMode(Field.Mode.NULLABLE)
            .build());
    fields.add(
        Field.newBuilder("SQL_DATETIME_SUB", StandardSQLTypeName.INT64)
            .setMode(Field.Mode.NULLABLE)
            .build());
    fields.add(
        Field.newBuilder("CHAR_OCTET_LENGTH", StandardSQLTypeName.INT64)
            .setMode(Field.Mode.NULLABLE)
            .build());
    fields.add(
        Field.newBuilder("ORDINAL_POSITION", StandardSQLTypeName.INT64)
            .setMode(Field.Mode.REQUIRED)
            .build());
    fields.add(
        Field.newBuilder("IS_NULLABLE", StandardSQLTypeName.STRING)
            .setMode(Field.Mode.REQUIRED)
            .build());
    fields.add(
        Field.newBuilder("SCOPE_CATALOG", StandardSQLTypeName.STRING)
            .setMode(Field.Mode.NULLABLE)
            .build());
    fields.add(
        Field.newBuilder("SCOPE_SCHEMA", StandardSQLTypeName.STRING)
            .setMode(Field.Mode.NULLABLE)
            .build());
    fields.add(
        Field.newBuilder("SCOPE_TABLE", StandardSQLTypeName.STRING)
            .setMode(Field.Mode.NULLABLE)
            .build());
    fields.add(
        Field.newBuilder("SOURCE_DATA_TYPE", StandardSQLTypeName.INT64)
            .setMode(Field.Mode.NULLABLE)
            .build());
    return Schema.of(fields);
  }

  @Override
  public boolean supportsResultSetHoldability(int holdability) {
    if (holdability == ResultSet.CLOSE_CURSORS_AT_COMMIT) {
      return true;
    }
    return false;
  }

  @Override
  public int getResultSetHoldability() {
    return ResultSet.CLOSE_CURSORS_AT_COMMIT;
  }

  @Override
  // Obtained from java libraries pom
  // https://github.com/googleapis/java-bigquery/blob/main/pom.xml
  public int getDatabaseMajorVersion() {
    return 2;
  }

  @Override
  public int getDatabaseMinorVersion() {
    return 0;
  }

  @Override
  public int getJDBCMajorVersion() {
    return 4;
  }

  @Override
  public int getJDBCMinorVersion() {
    return 2;
  }

  @Override
  public int getSQLStateType() {
    return DatabaseMetaData.sqlStateSQL;
  }

  @Override
  public boolean locatorsUpdateCopy() {
    return false;
  }

  @Override
  public boolean supportsStatementPooling() {
    return false;
  }

  @Override
  public RowIdLifetime getRowIdLifetime() {
    return null;
  }

  @Override
  public ResultSet getSchemas(String catalog, String schemaPattern) throws SQLException {
    return withTracing(
        "BigQueryDatabaseMetaData.getSchemas", () -> getSchemasImpl(catalog, schemaPattern));
  }

  private ResultSet getSchemasImpl(String catalog, String schemaPattern) throws SQLException {
    LOG.info("getSchemas called for catalog: %s, schemaPattern: %s", catalog, schemaPattern);
    final Schema resultSchema = defineGetSchemasSchema();
    if ((catalog != null && catalog.isEmpty())
        || (schemaPattern != null && schemaPattern.isEmpty())) {
      LOG.warning(
          "Returning empty ResultSet as one or more patterns are empty or catalog is empty.");
      return new BigQueryJsonResultSet(resultSchema);
    }

    final FieldList resultSchemaFields = resultSchema.getFields();
    final Pattern schemaRegex = compileSqlLikePattern(schemaPattern);
    BlockingQueue<BigQueryFieldValueListWrapper> queue =
        fetchAndPopulateQueue(
            () -> {
              List<Dataset> datasets = fetchMatchingDatasets(catalog, schemaPattern, schemaRegex);
              List<FieldValueList> collectedResults = new ArrayList<>(datasets.size());
              for (Dataset dataset : datasets) {
                processSchemaInfo(dataset.getDatasetId(), collectedResults, resultSchemaFields);
              }
              Comparator<FieldValueList> comparator =
                  defineGetSchemasComparator(resultSchemaFields);
              sortResults(collectedResults, comparator, "getSchemas", LOG);
              return collectedResults;
            },
            resultSchemaFields);
    return BigQueryJsonResultSet.of(resultSchema, -1, queue, null);
  }

  Schema defineGetSchemasSchema() {
    List<Field> fields = new ArrayList<>(2);
    fields.add(
        Field.newBuilder("TABLE_SCHEM", StandardSQLTypeName.STRING)
            .setMode(Field.Mode.REQUIRED)
            .build());
    fields.add(
        Field.newBuilder("TABLE_CATALOG", StandardSQLTypeName.STRING)
            .setMode(Field.Mode.NULLABLE)
            .build());
    return Schema.of(fields);
  }

  void processSchemaInfo(
      DatasetId datasetId, List<FieldValueList> collectedResults, FieldList resultSchemaFields) {
    LOG.finer("Processing schema info for dataset: " + datasetId);
    try {
      String schemaName = datasetId.getDataset();
      String catalogName = datasetId.getProject();
      List<FieldValue> values = new ArrayList<>(resultSchemaFields.size());
      values.add(createStringFieldValue(schemaName));
      values.add(createStringFieldValue(catalogName));
      FieldValueList rowFvl = FieldValueList.of(values, resultSchemaFields);
      collectedResults.add(rowFvl);
      LOG.finer("Processed and added schema info row for: " + datasetId);
    } catch (Exception e) {
      LOG.warning(
          "Error processing schema info for dataset %s: %s. Skipping this schema.",
          datasetId, e.getMessage());
    }
  }

  Comparator<FieldValueList> defineGetSchemasComparator(FieldList resultSchemaFields) {
    final int TABLE_CATALOG_IDX = resultSchemaFields.getIndex("TABLE_CATALOG");
    final int TABLE_SCHEM_IDX = resultSchemaFields.getIndex("TABLE_SCHEM");
    return Comparator.comparing(
            (FieldValueList fvl) -> getStringValueOrNull(fvl, TABLE_CATALOG_IDX),
            Comparator.nullsFirst(String::compareTo))
        .thenComparing(
            (FieldValueList fvl) -> getStringValueOrNull(fvl, TABLE_SCHEM_IDX),
            Comparator.nullsFirst(String::compareTo));
  }

  @Override
  public boolean supportsStoredFunctionsUsingCallSyntax() {
    return false;
  }

  @Override
  public boolean autoCommitFailureClosesAllResultSets() {
    return false;
  }

  @Override
  public ResultSet getClientInfoProperties() {
    LOG.info("getClientInfoProperties() called.");

    final Schema resultSchema = defineGetClientInfoPropertiesSchema();
    final FieldList resultSchemaFields = resultSchema.getFields();
    final BlockingQueue<BigQueryFieldValueListWrapper> queue = new LinkedBlockingQueue<>(4);
    final List<FieldValueList> collectedResults = new ArrayList<>(3);

    try {
      collectedResults.add(
          FieldValueList.of(
              Arrays.asList(
                  createStringFieldValue("ApplicationName"),
                  createLongFieldValue(25L),
                  createNullFieldValue(),
                  createStringFieldValue(
                      "The name of the application currently utilizing the connection.")),
              resultSchemaFields));

      collectedResults.add(
          FieldValueList.of(
              Arrays.asList(
                  createStringFieldValue("ClientHostname"),
                  createLongFieldValue(25L),
                  createNullFieldValue(),
                  createStringFieldValue(
                      "The hostname of the computer the application using the connection is running"
                          + " on.")),
              resultSchemaFields));

      collectedResults.add(
          FieldValueList.of(
              Arrays.asList(
                  createStringFieldValue("ClientUser"),
                  createLongFieldValue(25L),
                  createNullFieldValue(),
                  createStringFieldValue(
                      "The name of the user that the application using the connection is performing"
                          + " work for.")),
              resultSchemaFields));

      Comparator<FieldValueList> comparator =
          Comparator.comparing(
              (FieldValueList fvl) ->
                  getStringValueOrNull(fvl, resultSchemaFields.getIndex("NAME")),
              Comparator.nullsFirst(String::compareToIgnoreCase));
      sortResults(collectedResults, comparator, "getClientInfoProperties", LOG);
      populateQueue(collectedResults, queue, resultSchemaFields);

    } catch (Exception e) {
      LOG.warning("Unexpected error processing client info properties: " + e.getMessage());
      collectedResults.clear();
      queue.clear();
    } finally {
      signalEndOfData(queue, resultSchemaFields);
    }
    return BigQueryJsonResultSet.of(
        resultSchema, collectedResults.size(), queue, null, new Future<?>[0]);
  }

  Schema defineGetClientInfoPropertiesSchema() {
    List<Field> fields = new ArrayList<>(4);
    fields.add(
        Field.newBuilder("NAME", StandardSQLTypeName.STRING)
            .setMode(Field.Mode.REQUIRED)
            .build()); // 1
    fields.add(
        Field.newBuilder("MAX_LEN", StandardSQLTypeName.INT64)
            .setMode(Field.Mode.NULLABLE)
            .build()); // 2
    fields.add(
        Field.newBuilder("DEFAULT_VALUE", StandardSQLTypeName.STRING)
            .setMode(Field.Mode.NULLABLE)
            .build()); // 3
    fields.add(
        Field.newBuilder("DESCRIPTION", StandardSQLTypeName.STRING)
            .setMode(Field.Mode.NULLABLE)
            .build()); // 4
    return Schema.of(fields);
  }

  @Override
  public ResultSet getFunctions(String catalog, String schemaPattern, String functionNamePattern)
      throws SQLException {
    LOG.info(
        "getFunctions called for catalog: %s, schemaPattern: %s, functionNamePattern: %s",
        catalog, schemaPattern, functionNamePattern);

    final Schema resultSchema = defineGetFunctionsSchema();
    if ((catalog != null && catalog.isEmpty())
        || (schemaPattern != null && schemaPattern.isEmpty())
        || (functionNamePattern != null && functionNamePattern.isEmpty())) {
      LOG.warning(
          "Returning empty ResultSet as one or more patterns are empty or catalog is empty.");
      return new BigQueryJsonResultSet(resultSchema);
    }

    final FieldList resultSchemaFields = resultSchema.getFields();
    List<DatasetId> targetDatasets = getTargetDatasets(catalog, schemaPattern);

    BlockingQueue<BigQueryFieldValueListWrapper> queue =
        fetchAndPopulateQueue(
            () -> {
              final List<FieldValueList> collectedResults =
                  Collections.synchronizedList(new ArrayList<>());
              processTargetRoutinesConcurrently(
                  targetDatasets,
                  functionNamePattern,
                  true,
                  true,
                  collectedResults,
                  resultSchemaFields,
                  (routine, results, fields) -> {
                    if (!"SCALAR_FUNCTION".equalsIgnoreCase(routine.getRoutineType())
                        && !"TABLE_FUNCTION".equalsIgnoreCase(routine.getRoutineType())) {
                      return;
                    }
                    processFunctionInfo(routine, results, fields);
                  },
                  RoutineOption.fields(
                      RoutineField.ROUTINE_REFERENCE,
                      RoutineField.ARGUMENTS,
                      RoutineField.ROUTINE_TYPE));

              Comparator<FieldValueList> comparator =
                  defineGetFunctionsComparator(resultSchemaFields);
              sortResults(collectedResults, comparator, "getFunctions", LOG);
              return collectedResults;
            },
            resultSchemaFields);
    return BigQueryJsonResultSet.of(resultSchema, -1, queue, null);
  }

  Schema defineGetFunctionsSchema() {
    List<Field> fields = new ArrayList<>(6);
    fields.add(
        Field.newBuilder("FUNCTION_CAT", StandardSQLTypeName.STRING)
            .setMode(Mode.NULLABLE)
            .build());
    fields.add(
        Field.newBuilder("FUNCTION_SCHEM", StandardSQLTypeName.STRING)
            .setMode(Mode.NULLABLE)
            .build());
    fields.add(
        Field.newBuilder("FUNCTION_NAME", StandardSQLTypeName.STRING)
            .setMode(Mode.REQUIRED)
            .build());
    fields.add(
        Field.newBuilder("REMARKS", StandardSQLTypeName.STRING).setMode(Mode.NULLABLE).build());
    fields.add(
        Field.newBuilder("FUNCTION_TYPE", StandardSQLTypeName.INT64)
            .setMode(Mode.REQUIRED)
            .build());
    fields.add(
        Field.newBuilder("SPECIFIC_NAME", StandardSQLTypeName.STRING)
            .setMode(Mode.REQUIRED)
            .build());
    return Schema.of(fields);
  }

  void processFunctionInfo(
      Routine routine, List<FieldValueList> collectedResults, FieldList resultSchemaFields) {
    RoutineId routineId = routine.getRoutineId();
    LOG.fine("Processing function info for: " + routineId);

    try {
      String catalogName = routineId.getProject();
      String schemaName = routineId.getDataset();
      String functionName = routineId.getRoutine();
      String remarks = routine.getDescription();
      String specificName = functionName;

      short functionType;
      String bqRoutineType = routine.getRoutineType();
      if ("SCALAR_FUNCTION".equalsIgnoreCase(bqRoutineType)) {
        functionType = DatabaseMetaData.functionResultUnknown;
      } else if ("TABLE_FUNCTION".equalsIgnoreCase(bqRoutineType)) {
        functionType = DatabaseMetaData.functionReturnsTable;
      } else {
        functionType = DatabaseMetaData.functionResultUnknown;
      }

      List<FieldValue> values = new ArrayList<>(resultSchemaFields.size());
      values.add(createStringFieldValue(catalogName)); // 1. FUNCTION_CAT
      values.add(createStringFieldValue(schemaName)); // 2. FUNCTION_SCHEM
      values.add(createStringFieldValue(functionName)); // 3. FUNCTION_NAME
      values.add(createStringFieldValue(remarks)); // 4. REMARKS
      values.add(createLongFieldValue((long) functionType)); // 5. FUNCTION_TYPE
      values.add(createStringFieldValue(specificName)); // 6. SPECIFIC_NAME

      FieldValueList rowFvl = FieldValueList.of(values, resultSchemaFields);
      collectedResults.add(rowFvl);
      LOG.fine("Processed and added function info row for: " + routineId);

    } catch (Exception e) {
      LOG.warning(
          "Error processing function info for %s: %s. Skipping this function.",
          routineId, e.getMessage());
    }
  }

  Comparator<FieldValueList> defineGetFunctionsComparator(FieldList resultSchemaFields) {
    final int FUNC_CAT_IDX = resultSchemaFields.getIndex("FUNCTION_CAT");
    final int FUNC_SCHEM_IDX = resultSchemaFields.getIndex("FUNCTION_SCHEM");
    final int FUNC_NAME_IDX = resultSchemaFields.getIndex("FUNCTION_NAME");
    final int SPEC_NAME_IDX = resultSchemaFields.getIndex("SPECIFIC_NAME");

    return Comparator.comparing(
            (FieldValueList fvl) -> getStringValueOrNull(fvl, FUNC_CAT_IDX),
            Comparator.nullsFirst(String::compareTo))
        .thenComparing(
            (FieldValueList fvl) -> getStringValueOrNull(fvl, FUNC_SCHEM_IDX),
            Comparator.nullsFirst(String::compareTo))
        .thenComparing(
            (FieldValueList fvl) -> getStringValueOrNull(fvl, FUNC_NAME_IDX),
            Comparator.nullsFirst(String::compareTo))
        .thenComparing(
            (FieldValueList fvl) -> getStringValueOrNull(fvl, SPEC_NAME_IDX),
            Comparator.nullsFirst(String::compareTo));
  }

  @Override
  public ResultSet getFunctionColumns(
      String catalog, String schemaPattern, String functionNamePattern, String columnNamePattern)
      throws SQLException {
    LOG.info(
        "getFunctionColumns called for catalog: %s, schemaPattern: %s, functionNamePattern: %s, columnNamePattern: %s",
        catalog, schemaPattern, functionNamePattern, columnNamePattern);

    final Schema resultSchema = defineGetFunctionColumnsSchema();
    if ((catalog != null && catalog.isEmpty())
        || (schemaPattern != null && schemaPattern.isEmpty())
        || (functionNamePattern != null && functionNamePattern.isEmpty())
        || (columnNamePattern != null && columnNamePattern.isEmpty())) {
      LOG.warning(
          "Returning empty ResultSet as one or more patterns are empty or catalog is empty.");
      return new BigQueryJsonResultSet(resultSchema);
    }

    final FieldList resultSchemaFields = resultSchema.getFields();
    List<DatasetId> targetDatasets = getTargetDatasets(catalog, schemaPattern);

    Pattern columnNameRegex = compileSqlLikePattern(columnNamePattern);

    BlockingQueue<BigQueryFieldValueListWrapper> queue =
        fetchAndPopulateQueue(
            () -> {
              final List<FieldValueList> collectedResults =
                  Collections.synchronizedList(new ArrayList<>());
              processTargetRoutinesConcurrently(
                  targetDatasets,
                  functionNamePattern,
                  true,
                  true,
                  collectedResults,
                  resultSchemaFields,
                  (routine, results, fields) -> {
                    if (!"SCALAR_FUNCTION".equalsIgnoreCase(routine.getRoutineType())
                        && !"TABLE_FUNCTION".equalsIgnoreCase(routine.getRoutineType())) {
                      return;
                    }
                    processFunctionParametersAndReturnValue(
                        routine, columnNameRegex, results, fields);
                  },
                  RoutineOption.fields(
                      RoutineField.ROUTINE_REFERENCE,
                      RoutineField.ARGUMENTS,
                      RoutineField.ROUTINE_TYPE));

              Comparator<FieldValueList> comparator =
                  defineGetFunctionColumnsComparator(resultSchemaFields);
              sortResults(collectedResults, comparator, "getFunctionColumns", LOG);
              return collectedResults;
            },
            resultSchemaFields);
    return BigQueryJsonResultSet.of(resultSchema, -1, queue, null);
  }

  Schema defineGetFunctionColumnsSchema() {
    List<Field> fields = new ArrayList<>(17);
    fields.add(
        Field.newBuilder("FUNCTION_CAT", StandardSQLTypeName.STRING)
            .setMode(Mode.NULLABLE)
            .build()); // 1
    fields.add(
        Field.newBuilder("FUNCTION_SCHEM", StandardSQLTypeName.STRING)
            .setMode(Mode.NULLABLE)
            .build()); // 2
    fields.add(
        Field.newBuilder("FUNCTION_NAME", StandardSQLTypeName.STRING)
            .setMode(Mode.REQUIRED)
            .build()); // 3
    fields.add(
        Field.newBuilder("COLUMN_NAME", StandardSQLTypeName.STRING)
            .setMode(Mode.REQUIRED)
            .build()); // 4
    fields.add(
        Field.newBuilder("COLUMN_TYPE", StandardSQLTypeName.INT64)
            .setMode(Mode.REQUIRED)
            .build()); // 5
    fields.add(
        Field.newBuilder("DATA_TYPE", StandardSQLTypeName.INT64)
            .setMode(Mode.REQUIRED)
            .build()); // 6
    fields.add(
        Field.newBuilder("TYPE_NAME", StandardSQLTypeName.STRING)
            .setMode(Mode.REQUIRED)
            .build()); // 7
    fields.add(
        Field.newBuilder("PRECISION", StandardSQLTypeName.INT64)
            .setMode(Mode.NULLABLE)
            .build()); // 8
    fields.add(
        Field.newBuilder("LENGTH", StandardSQLTypeName.INT64).setMode(Mode.NULLABLE).build()); // 9
    fields.add(
        Field.newBuilder("SCALE", StandardSQLTypeName.INT64).setMode(Mode.NULLABLE).build()); // 10
    fields.add(
        Field.newBuilder("RADIX", StandardSQLTypeName.INT64).setMode(Mode.NULLABLE).build()); // 11
    fields.add(
        Field.newBuilder("NULLABLE", StandardSQLTypeName.INT64)
            .setMode(Mode.REQUIRED)
            .build()); // 12
    fields.add(
        Field.newBuilder("REMARKS", StandardSQLTypeName.STRING)
            .setMode(Mode.NULLABLE)
            .build()); // 13
    fields.add(
        Field.newBuilder("CHAR_OCTET_LENGTH", StandardSQLTypeName.INT64)
            .setMode(Mode.NULLABLE)
            .build()); // 14
    fields.add(
        Field.newBuilder("ORDINAL_POSITION", StandardSQLTypeName.INT64)
            .setMode(Mode.REQUIRED)
            .build()); // 15
    fields.add(
        Field.newBuilder("IS_NULLABLE", StandardSQLTypeName.STRING)
            .setMode(Mode.REQUIRED)
            .build()); // 16
    fields.add(
        Field.newBuilder("SPECIFIC_NAME", StandardSQLTypeName.STRING)
            .setMode(Mode.REQUIRED)
            .build()); // 17
    return Schema.of(fields);
  }

  void processFunctionParametersAndReturnValue(
      Routine routine,
      Pattern columnNameRegex,
      List<FieldValueList> collectedResults,
      FieldList resultSchemaFields) {
    RoutineId routineId = routine.getRoutineId();
    if (routineId == null) {
      LOG.warning("Processing a routine with a null ID. Skipping.");
      return;
    }
    LOG.finer("Processing function parameters and return value for: " + routineId);

    String functionCatalog = routineId.getProject();
    String functionSchema = routineId.getDataset();
    String functionName = routineId.getRoutine();
    String specificName = functionName;

    if (routine.getReturnTableType() != null) {
      StandardSQLTableType returnTableType = routine.getReturnTableType();
      if (returnTableType != null && returnTableType.getColumns() != null) {
        List<StandardSQLField> tableColumns = returnTableType.getColumns();
        for (int i = 0; i < tableColumns.size(); i++) {
          StandardSQLField tableColumn = tableColumns.get(i);
          String columnName = tableColumn.getName();
          if (columnNameRegex != null
              && (columnName == null || !columnNameRegex.matcher(columnName).matches())) {
            continue;
          }
          List<FieldValue> rowValues =
              createFunctionColumnRow(
                  functionCatalog,
                  functionSchema,
                  functionName,
                  specificName,
                  columnName,
                  DatabaseMetaData.functionColumnResult,
                  tableColumn.getDataType(),
                  i + 1);
          collectedResults.add(FieldValueList.of(rowValues, resultSchemaFields));
        }
      }
    }

    List<RoutineArgument> arguments = routine.getArguments();
    if (arguments != null) {
      for (int i = 0; i < arguments.size(); i++) {
        RoutineArgument arg = arguments.get(i);
        String argName = arg.getName();

        if (columnNameRegex != null
            && (argName == null || !columnNameRegex.matcher(argName).matches())) {
          continue;
        }

        short columnType;
        String originalMode = arg.getMode();

        if ("IN".equalsIgnoreCase(originalMode)) {
          columnType = DatabaseMetaData.functionColumnIn;
        } else if ("OUT".equalsIgnoreCase(originalMode)) {
          columnType = DatabaseMetaData.functionColumnOut;
        } else if ("INOUT".equalsIgnoreCase(originalMode)) {
          columnType = DatabaseMetaData.functionColumnInOut;
        } else {
          columnType = DatabaseMetaData.functionColumnUnknown;
        }

        List<FieldValue> rowValues =
            createFunctionColumnRow(
                functionCatalog,
                functionSchema,
                functionName,
                specificName,
                argName,
                columnType,
                arg.getDataType(),
                i + 1);
        collectedResults.add(FieldValueList.of(rowValues, resultSchemaFields));
      }
    }
  }

  List<FieldValue> createFunctionColumnRow(
      String functionCatalog,
      String functionSchema,
      String functionName,
      String specificName,
      String columnName,
      int columnType,
      StandardSQLDataType dataType,
      int ordinalPosition) {

    List<FieldValue> values = new ArrayList<>(17);
    ColumnTypeInfo typeInfo =
        determineTypeInfoFromDataType(dataType, functionName, columnName, ordinalPosition);

    values.add(createStringFieldValue(functionCatalog)); // 1. FUNCTION_CAT
    values.add(createStringFieldValue(functionSchema)); // 2. FUNCTION_SCHEM
    values.add(createStringFieldValue(functionName)); // 3. FUNCTION_NAME
    values.add(createStringFieldValue(columnName)); // 4. COLUMN_NAME
    values.add(createLongFieldValue((long) columnType)); // 5. COLUMN_TYPE

    values.add(createLongFieldValue((long) typeInfo.jdbcType)); // 6. DATA_TYPE
    values.add(createStringFieldValue(typeInfo.typeName)); // 7. TYPE_NAME
    values.add(
        createLongFieldValue(
            typeInfo.columnSize == null ? null : typeInfo.columnSize.longValue())); // 8. PRECISION
    if (typeInfo.typeName != null
        && (typeInfo.typeName.equalsIgnoreCase("STRING")
            || typeInfo.typeName.equalsIgnoreCase("NVARCHAR")
            || typeInfo.typeName.equalsIgnoreCase("BYTES")
            || typeInfo.typeName.equalsIgnoreCase("VARBINARY"))) {
      values.add(
          createLongFieldValue(
              typeInfo.columnSize == null ? null : typeInfo.columnSize.longValue())); // 9. LENGTH
      values.add(
          createLongFieldValue(
              typeInfo.decimalDigits == null
                  ? null
                  : typeInfo.decimalDigits.longValue())); // 10. SCALE
      values.add(
          createLongFieldValue(
              typeInfo.numPrecRadix == null
                  ? null
                  : typeInfo.numPrecRadix.longValue())); // 11. RADIX
      values.add(
          createLongFieldValue((long) DatabaseMetaData.functionNullableUnknown)); // 12. NULLABLE
      values.add(createStringFieldValue(null)); // 13. REMARKS
      values.add(
          createLongFieldValue(
              typeInfo.columnSize == null
                  ? null
                  : typeInfo.columnSize.longValue())); // 14. CHAR_OCTET_LENGTH
    } else {
      values.add(createNullFieldValue()); // 9. LENGTH
      values.add(
          createLongFieldValue(
              typeInfo.decimalDigits == null
                  ? null
                  : typeInfo.decimalDigits.longValue())); // 10. SCALE
      values.add(
          createLongFieldValue(
              typeInfo.numPrecRadix == null
                  ? null
                  : typeInfo.numPrecRadix.longValue())); // 11. RADIX
      values.add(
          createLongFieldValue((long) DatabaseMetaData.functionNullableUnknown)); // 12. NULLABLE
      values.add(createStringFieldValue(null)); // 13. REMARKS
      values.add(createNullFieldValue()); // 14. CHAR_OCTET_LENGTH
    }

    values.add(createLongFieldValue((long) ordinalPosition)); // 15. ORDINAL_POSITION
    values.add(createStringFieldValue("")); // 16. IS_NULLABLE
    values.add(createStringFieldValue(specificName)); // 17. SPECIFIC_NAME

    return values;
  }

  Comparator<FieldValueList> defineGetFunctionColumnsComparator(FieldList resultSchemaFields) {
    final int FUNC_CAT_IDX = resultSchemaFields.getIndex("FUNCTION_CAT");
    final int FUNC_SCHEM_IDX = resultSchemaFields.getIndex("FUNCTION_SCHEM");
    final int FUNC_NAME_IDX = resultSchemaFields.getIndex("FUNCTION_NAME");
    final int SPEC_NAME_IDX = resultSchemaFields.getIndex("SPECIFIC_NAME");
    final int ORDINAL_POS_IDX = resultSchemaFields.getIndex("ORDINAL_POSITION");

    return Comparator.comparing(
            (FieldValueList fvl) -> getStringValueOrNull(fvl, FUNC_CAT_IDX),
            Comparator.nullsFirst(String::compareToIgnoreCase))
        .thenComparing(
            (FieldValueList fvl) -> getStringValueOrNull(fvl, FUNC_SCHEM_IDX),
            Comparator.nullsFirst(String::compareToIgnoreCase))
        .thenComparing(
            (FieldValueList fvl) -> getStringValueOrNull(fvl, FUNC_NAME_IDX),
            Comparator.nullsFirst(String::compareToIgnoreCase))
        .thenComparing(
            (FieldValueList fvl) -> getStringValueOrNull(fvl, SPEC_NAME_IDX),
            Comparator.nullsFirst(String::compareToIgnoreCase))
        .thenComparing(
            (FieldValueList fvl) -> getLongValueOrNull(fvl, ORDINAL_POS_IDX),
            Comparator.nullsFirst(Long::compareTo));
  }

  @Override
  public ResultSet getPseudoColumns(
      String catalog, String schemaPattern, String tableNamePattern, String columnNamePattern) {
    LOG.info(
        "getPseudoColumns called for catalog: %s, schemaPattern: %s, tableNamePattern: %s,"
            + " columnNamePattern: %s. Pseudo columns not supported by BigQuery; returning"
            + " empty ResultSet.",
        catalog, schemaPattern, tableNamePattern, columnNamePattern);

    return new BigQueryJsonResultSet(defineGetPseudoColumnsSchema());
  }

  Schema defineGetPseudoColumnsSchema() {
    List<Field> fields = new ArrayList<>(12);
    fields.add(
        Field.newBuilder("TABLE_CAT", StandardSQLTypeName.STRING)
            .setMode(Field.Mode.NULLABLE)
            .build());
    fields.add(
        Field.newBuilder("TABLE_SCHEM", StandardSQLTypeName.STRING)
            .setMode(Field.Mode.NULLABLE)
            .build());
    fields.add(
        Field.newBuilder("TABLE_NAME", StandardSQLTypeName.STRING)
            .setMode(Field.Mode.REQUIRED)
            .build());
    fields.add(
        Field.newBuilder("COLUMN_NAME", StandardSQLTypeName.STRING)
            .setMode(Field.Mode.REQUIRED)
            .build());
    fields.add(
        Field.newBuilder("DATA_TYPE", StandardSQLTypeName.INT64)
            .setMode(Field.Mode.REQUIRED)
            .build());
    fields.add(
        Field.newBuilder("COLUMN_SIZE", StandardSQLTypeName.INT64)
            .setMode(Field.Mode.REQUIRED)
            .build());
    fields.add(
        Field.newBuilder("DECIMAL_DIGITS", StandardSQLTypeName.INT64)
            .setMode(Field.Mode.NULLABLE)
            .build());
    fields.add(
        Field.newBuilder("NUM_PREC_RADIX", StandardSQLTypeName.INT64)
            .setMode(Field.Mode.NULLABLE)
            .build());
    fields.add(
        Field.newBuilder("COLUMN_USAGE", StandardSQLTypeName.STRING)
            .setMode(Field.Mode.REQUIRED)
            .build());
    fields.add(
        Field.newBuilder("REMARKS", StandardSQLTypeName.STRING)
            .setMode(Field.Mode.NULLABLE)
            .build());
    fields.add(
        Field.newBuilder("CHAR_OCTET_LENGTH", StandardSQLTypeName.INT64)
            .setMode(Field.Mode.NULLABLE)
            .build());
    fields.add(
        Field.newBuilder("IS_NULLABLE", StandardSQLTypeName.STRING)
            .setMode(Field.Mode.REQUIRED)
            .build());
    return Schema.of(fields);
  }

  @Override
  public boolean generatedKeyAlwaysReturned() {
    return false;
  }

  @Override
  public <T> T unwrap(Class<T> iface) throws SQLException {
    if (iface.isInstance(this)) {
      return iface.cast(this);
    }
    throw new BigQueryJdbcException("Cannot unwrap to " + iface.getName());
  }

  @Override
  public boolean isWrapperFor(Class<?> iface) throws SQLException {
    return iface != null && iface.isInstance(this);
  }

  // --- Helper Methods ---

  /**
   * Determines the effective catalog and schema pattern to use for metadata retrieval.
   *
   * <p>This method applies the logic for the {@code FilterTablesOnDefaultDataset} connection
   * property. If this property is enabled and the provided {@code catalog} or {@code schemaPattern}
   * are null, empty, or wildcard ('%'), they may be overridden by the default catalog (project) and
   * default dataset (schema) configured in the {@link BigQueryConnection}.
   *
   * @param catalog The catalog name provided by the user; may be {@code null}.
   * @param schemaPattern The schema name pattern provided by the user; may be {@code null}.
   * @return A {@link Tuple} where {@code Tuple.x()} is the effective catalog string and {@code
   *     Tuple.y()} is the effective schema pattern string. These are the values that should be used
   *     for querying BigQuery's metadata.
   * @see BigQueryConnection#isFilterTablesOnDefaultDataset()
   */
  Tuple<String, String> determineEffectiveCatalogAndSchema(String catalog, String schemaPattern)
      throws BigQueryJdbcException {
    String effectiveCatalog = catalog;
    String effectiveSchemaPattern = schemaPattern;

    if (this.connection.isFilterTablesOnDefaultDataset()
        && this.connection.getDefaultDataset() != null
        && this.connection.getDefaultDataset().getDataset() != null
        && !this.connection.getDefaultDataset().getDataset().isEmpty()) {

      String defaultProjectFromConnection =
          (this.connection.getDefaultDataset().getProject() != null)
              ? this.connection.getDefaultDataset().getProject()
              : this.connection.getCatalog();
      // We only use the dataset part of the DefaultDataset for schema filtering
      String defaultSchemaFromConnection = this.connection.getDefaultDataset().getDataset();

      boolean catalogIsUnspecified = (effectiveCatalog == null || effectiveCatalog.equals("%"));
      boolean schemaIsUnspecified =
          (effectiveSchemaPattern == null || effectiveSchemaPattern.equals("%"));

      final String logPrefix = "FilterTablesOnDefaultDatasetTrue: ";
      if (catalogIsUnspecified && schemaIsUnspecified) {
        effectiveCatalog = defaultProjectFromConnection;
        effectiveSchemaPattern = defaultSchemaFromConnection;
        LOG.info(
            logPrefix + "Using default catalog '%s' and default dataset '%s'.",
            effectiveCatalog,
            effectiveSchemaPattern);
      } else if (catalogIsUnspecified) {
        effectiveCatalog = null;
        LOG.info(
            logPrefix + "Using all catalogs with user dataset '%s'. Default dataset '%s' ignored.",
            effectiveSchemaPattern,
            defaultSchemaFromConnection);
      } else if (schemaIsUnspecified) {
        effectiveSchemaPattern = defaultSchemaFromConnection;
        LOG.info(
            logPrefix + "Using user catalog '%s' and default dataset '%s'.",
            effectiveCatalog,
            effectiveSchemaPattern);
      } else {
        LOG.info(
            logPrefix + "Using user catalog '%s' and schema '%s'. Default dataset '%s' ignored.",
            effectiveCatalog,
            effectiveSchemaPattern,
            defaultSchemaFromConnection);
      }
    }
    return Tuple.of(effectiveCatalog, effectiveSchemaPattern);
  }

  private ColumnTypeInfo getColumnTypeInfoForSqlType(StandardSQLTypeName bqType) {
    if (bqType == null) {
      LOG.warning("Null BigQuery type encountered. Mapping to STRING.");
      return new ColumnTypeInfo(Types.NVARCHAR, "STRING", null, null, null);
    }
    ColumnTypeInfo info = BigQueryTypeRegistry.getColumnTypeInfo(bqType);
    if (info == null) {
      LOG.warning("Unknown BigQuery type encountered: " + bqType.name() + ". Mapping to STRING.");
      return new ColumnTypeInfo(Types.NVARCHAR, "STRING", null, null, null);
    }
    return info;
  }

  <T> List<T> findMatchingBigQueryObjects(
      String objectTypeName,
      Supplier<Page<T>> listAllOperation,
      Function<String, T> getSpecificOperation,
      Function<T, String> nameExtractor,
      String pattern,
      Pattern regex,
      boolean throwOn404)
      throws InterruptedException {

    boolean needsList = needsListing(pattern);
    List<T> resultList = new ArrayList<>();

    try {
      Iterable<T> objects;
      if (needsList) {
        LOG.info(
            "Listing all %ss (pattern: %s)...",
            objectTypeName, pattern == null ? "<all>" : pattern);
        Page<T> firstPage = listAllOperation.get();
        objects = firstPage.iterateAll();
        LOG.fine("Retrieved initial %s list, iterating & filtering if needed...", objectTypeName);

      } else {
        LOG.info("Getting specific %s: '%s'", objectTypeName, pattern);
        T specificObject = getSpecificOperation.apply(pattern);
        objects =
            (specificObject == null)
                ? Collections.<T>emptyList()
                : Collections.singletonList(specificObject);
        if (specificObject == null) {
          LOG.info("Specific %s not found: '%s'", objectTypeName, pattern);
        }
      }

      boolean wasListing = needsList;
      for (T obj : objects) {
        if (Thread.currentThread().isInterrupted()) {
          LOG.warning("Thread interrupted during " + objectTypeName + " processing loop.");
          throw new InterruptedException(
              "Interrupted during " + objectTypeName + " processing loop");
        }
        if (obj != null) {
          if (wasListing && regex != null) {
            String name = nameExtractor.apply(obj);
            if (name != null && regex.matcher(name).matches()) {
              resultList.add(obj);
            }
          } else {
            resultList.add(obj);
          }
        }
      }

    } catch (BigQueryException e) {
      if (e.getCode() == 404 && !throwOn404) {
        LOG.info("%s '%s' not found (API error 404).", objectTypeName, pattern);
        return Collections.emptyList();
      } else {
        LOG.warning(
            "BigQueryException finding %ss for pattern '%s': %s (Code: %d)",
            objectTypeName, pattern, e.getMessage(), e.getCode());
        throw e;
      }
    } catch (InterruptedException e) {
      Thread.currentThread().interrupt();
      LOG.warning("Interrupted while finding " + objectTypeName + "s.");
      throw e;
    } catch (Exception e) {
      LOG.severe(
          "Unexpected exception finding %ss for pattern '%s': %s",
          objectTypeName, pattern, e.getMessage());
      throw new RuntimeException(e);
    }
    return resultList;
  }

  private static class TypeInfoRowData {
    String typeName;
    int jdbcType;
    Long precision;
    String literalPrefix;
    String literalSuffix;
    String createParams;
    int nullable;
    boolean caseSensitive;
    int searchable;
    boolean unsignedAttribute;
    boolean fixedPrecScale;
    boolean autoIncrement;
    String localTypeName;
    Long minimumScale;
    Long maximumScale;
    Long numPrecRadix;

    TypeInfoRowData(
        String typeName,
        int jdbcType,
        Long precision,
        String literalPrefix,
        String literalSuffix,
        String createParams,
        int nullable,
        boolean caseSensitive,
        int searchable,
        boolean unsignedAttribute,
        boolean fixedPrecScale,
        boolean autoIncrement,
        String localTypeName,
        Long minimumScale,
        Long maximumScale,
        Long numPrecRadix) {
      this.typeName = typeName;
      this.jdbcType = jdbcType;
      this.precision = precision;
      this.literalPrefix = literalPrefix;
      this.literalSuffix = literalSuffix;
      this.createParams = createParams;
      this.nullable = nullable;
      this.caseSensitive = caseSensitive;
      this.searchable = searchable;
      this.unsignedAttribute = unsignedAttribute;
      this.fixedPrecScale = fixedPrecScale;
      this.autoIncrement = autoIncrement;
      this.localTypeName = localTypeName;
      this.minimumScale = minimumScale;
      this.maximumScale = maximumScale;
      this.numPrecRadix = numPrecRadix;
    }
  }

  void sortResults(
      List<FieldValueList> collectedResults,
      Comparator<FieldValueList> comparator,
      String operationName,
      BigQueryJdbcCustomLogger logger) {

    if (collectedResults == null || collectedResults.isEmpty()) {
      logger.info("No results collected for %s, skipping sort.", operationName);
      return;
    }
    if (comparator == null) {
      logger.info("No comparator provided for %s, skipping sort.", operationName);
      return;
    }

    logger.info("Sorting %d collected %s results...", collectedResults.size(), operationName);
    try {
      collectedResults.sort(comparator);
      logger.info("%s result sorting completed.", operationName);
    } catch (Exception e) {
      logger.severe("Error during sorting %s results: %s", operationName, e.getMessage());
    }
  }

  private List<Field> defineBasePrivilegeFields() {
    List<Field> fields = new ArrayList<>(7);
    fields.add(
        Field.newBuilder("TABLE_CAT", StandardSQLTypeName.STRING)
            .setMode(Field.Mode.NULLABLE)
            .build());
    fields.add(
        Field.newBuilder("TABLE_SCHEM", StandardSQLTypeName.STRING)
            .setMode(Field.Mode.NULLABLE)
            .build());
    fields.add(
        Field.newBuilder("TABLE_NAME", StandardSQLTypeName.STRING)
            .setMode(Field.Mode.REQUIRED)
            .build());
    fields.add(
        Field.newBuilder("GRANTOR", StandardSQLTypeName.STRING)
            .setMode(Field.Mode.NULLABLE)
            .build());
    fields.add(
        Field.newBuilder("GRANTEE", StandardSQLTypeName.STRING)
            .setMode(Field.Mode.REQUIRED)
            .build());
    fields.add(
        Field.newBuilder("PRIVILEGE", StandardSQLTypeName.STRING)
            .setMode(Field.Mode.REQUIRED)
            .build());
    fields.add(
        Field.newBuilder("IS_GRANTABLE", StandardSQLTypeName.STRING)
            .setMode(Field.Mode.NULLABLE)
            .build());
    return fields;
  }

  Pattern compileSqlLikePattern(String sqlLikePattern) {
    if (sqlLikePattern == null) {
      return null;
    }
    if (sqlLikePattern.isEmpty()) {
      return Pattern.compile("(?!)");
    }
    StringBuilder regex = new StringBuilder(sqlLikePattern.length() * 2);
    regex.append('^');
    boolean escaped = false;
    for (int i = 0; i < sqlLikePattern.length(); i++) {
      char c = sqlLikePattern.charAt(i);
      if (!escaped && c == '\\') {
        escaped = true;
        continue;
      } else if (!escaped && c == '%') {
        regex.append(".*");
      } else if (!escaped && c == '_') {
        regex.append('.');
      } else {
        if (isRegexMetacharacter(c)) {
          regex.append('\\');
        }
        regex.append(c);
        escaped = false;
      }
    }
    if (escaped) {
      regex.append('\\').append('\\');
    }
    regex.append('$');
    return Pattern.compile(regex.toString(), Pattern.CASE_INSENSITIVE);
  }

  private static boolean isRegexMetacharacter(char c) {
    return "\\.[]{}()*+?^$|".indexOf(c) != -1;
  }

  boolean needsListing(String pattern) {
    return pattern == null || hasWildcards(pattern);
  }

  boolean hasWildcards(String pattern) {
    return pattern != null && (pattern.contains("%") || pattern.contains("_"));
  }

  FieldValue createStringFieldValue(String value) {
    return FieldValue.of(FieldValue.Attribute.PRIMITIVE, value);
  }

  FieldValue createLongFieldValue(Long value) {
    return (value == null)
        ? FieldValue.of(FieldValue.Attribute.PRIMITIVE, null)
        : FieldValue.of(FieldValue.Attribute.PRIMITIVE, String.valueOf(value));
  }

  FieldValue createNullFieldValue() {
    return FieldValue.of(FieldValue.Attribute.PRIMITIVE, null);
  }

  FieldValue createBooleanFieldValue(Boolean value) {
    return (value == null)
        ? FieldValue.of(FieldValue.Attribute.PRIMITIVE, null)
        : FieldValue.of(FieldValue.Attribute.PRIMITIVE, value ? "1" : "0");
  }

  private String getStringValueOrNull(FieldValueList fvl, int index) {
    if (fvl == null || index < 0 || index >= fvl.size()) return null;
    FieldValue fv = fvl.get(index);
    return (fv == null || fv.isNull()) ? null : fv.getStringValue();
  }

  private Long getLongValueOrNull(FieldValueList fvl, int index) {
    if (fvl == null || index < 0 || index >= fvl.size()) return null;
    FieldValue fv = fvl.get(index);
    try {
      return (fv == null || fv.isNull()) ? null : fv.getLongValue();
    } catch (NumberFormatException e) {
      LOG.warning("Could not parse Long value for index " + index);
      return null;
    }
  }

  private void waitForTasksCompletion(List<Future<?>> taskFutures)
      throws ExecutionException, SQLException {
    LOG.info("Waiting for %d submitted tasks to complete...", taskFutures.size());
    try {
      for (Future<?> future : taskFutures) {
        if (!future.isCancelled()) {
          future.get();
        }
      }
    } catch (CancellationException e) {
      LOG.warning("A table processing task was cancelled.");
    } catch (InterruptedException e) {
      Thread.currentThread().interrupt();
      LOG.warning(
          "Fetcher thread interrupted while waiting for tasks. Attempting to cancel remaining tasks.");
      throw new BigQueryJdbcException("Metadata fetch interrupted.", e);
    } finally {
      for (Future<?> future : taskFutures) {
        if (!future.isDone()) {
          future.cancel(true);
        }
      }
    }
    LOG.info("Finished waiting for tasks.");
  }

  private BlockingQueue<BigQueryFieldValueListWrapper> fetchAndPopulateQueue(
      Callable<List<FieldValueList>> task, FieldList resultSchemaFields) throws SQLException {
    BlockingQueue<BigQueryFieldValueListWrapper> queue = new LinkedBlockingQueue<>();
    try {
      List<FieldValueList> collectedResults = task.call();
      populateQueue(collectedResults, queue, resultSchemaFields);
      signalEndOfData(queue, resultSchemaFields);
      return queue;
    } catch (InterruptedException e) {
      Thread.currentThread().interrupt();
      throw new BigQueryJdbcException("Error during metadata fetch", e);
    } catch (Exception e) {
      if (e instanceof SQLException) {
        throw (SQLException) e;
      }
      throw new SQLException("Error during metadata fetch", e);
    }
  }

  private void populateQueue(
      List<FieldValueList> collectedResults,
      BlockingQueue<BigQueryFieldValueListWrapper> queue,
      FieldList resultSchemaFields) {
    LOG.info("Populating queue with %d results...", collectedResults.size());
    try {
      boolean[] isComplexColumn =
          BigQueryFieldValueListWrapper.createComplexColumnFlags(resultSchemaFields);
      for (FieldValueList sortedRow : collectedResults) {
        if (Thread.currentThread().isInterrupted()) {
          LOG.warning("Interrupted during queue population.");
          break;
        }
        queue.put(BigQueryFieldValueListWrapper.of(resultSchemaFields, sortedRow, isComplexColumn));
      }
      LOG.info("Finished populating queue.");
    } catch (InterruptedException e) {
      Thread.currentThread().interrupt();
      LOG.warning("Interrupted while putting row onto queue.");
    } catch (Exception e) {
      LOG.severe("Unexpected error populating queue: " + e.getMessage());
    }
  }

  private void signalEndOfData(
      BlockingQueue<BigQueryFieldValueListWrapper> queue, FieldList resultSchemaFields) {
    try {
      LOG.info("Adding end signal to queue.");
      BigQueryFieldValueListWrapper element =
          BigQueryFieldValueListWrapper.ofEndOfStream(resultSchemaFields);
      queue.add(element);
    } catch (Exception e) {
      LOG.severe("Exception while sending end signal to queue: " + e.getMessage());
    }
  }

  private List<Dataset> fetchDatasetsForProject(
      String project, String schemaPattern, Pattern schemaRegex, boolean throwOn404)
      throws SQLException {
    try {
      List<Dataset> datasets =
          findMatchingBigQueryObjects(
              "Dataset",
              () -> bigquery.listDatasets(project, DatasetListOption.pageSize(DEFAULT_PAGE_SIZE)),
              (name) -> bigquery.getDataset(DatasetId.of(project, name)),
              (ds) -> ds.getDatasetId().getDataset(),
              schemaPattern,
              schemaRegex,
              throwOn404);
      return datasets != null ? datasets : Collections.emptyList();
    } catch (InterruptedException e) {
      Thread.currentThread().interrupt();
      throw new BigQueryJdbcException(
          "Interrupted while fetching matching datasets for project " + project, e);
    } catch (Exception e) {
      throw new SQLException("Failed to fetch matching datasets for project " + project, e);
    }
  }

  private List<Dataset> fetchMatchingDatasets(
      String catalog, String schemaPattern, Pattern schemaRegex) throws SQLException {
    List<String> projects =
        (catalog != null) ? Collections.singletonList(catalog) : getAccessibleCatalogNames();

    if (projects.isEmpty()) {
      return Collections.emptyList();
    }

    boolean isBroadScan = (catalog == null);

    // Single project path
    if (projects.size() == 1) {
      return fetchDatasetsForProject(projects.get(0), schemaPattern, schemaRegex, isBroadScan);
    }

    // Multi-project path
    final List<Dataset> allDatasets = Collections.synchronizedList(new ArrayList<>());
    final List<Future<?>> taskFutures = new ArrayList<>();
    ExecutorService executor = connection.getMetadataExecutor();

    try {
      for (String project : projects) {
        Callable<Void> task =
            () -> {
              List<Dataset> datasets =
                  fetchDatasetsForProject(project, schemaPattern, schemaRegex, isBroadScan);
              allDatasets.addAll(datasets);
              return null;
            };
        taskFutures.add(executor.submit(task));
      }

      waitForTasksCompletion(taskFutures);
    } catch (ExecutionException e) {
      Throwable cause = e.getCause();
      if (cause instanceof SQLException) {
        throw (SQLException) cause;
      }
      throw new SQLException("Error parallel-fetching matching datasets", e);
    } finally {
      taskFutures.forEach(f -> f.cancel(true));
    }

    return allDatasets;
  }

  List<String> getAccessibleCatalogNames() throws SQLException {
    Set<String> accessibleCatalogs = new HashSet<>();
    String primaryCatalog = this.connection.getCatalog();
    if (primaryCatalog != null && !primaryCatalog.isEmpty()) {
      accessibleCatalogs.add(primaryCatalog);
    }

    if (this.connection.getDefaultDataset() != null
        && this.connection.getDefaultDataset().getProject() != null
        && !this.connection.getDefaultDataset().getProject().isEmpty()) {
      accessibleCatalogs.add(this.connection.getDefaultDataset().getProject());
    }

    String additionalProjectsStr = this.connection.getAdditionalProjects();
    if (additionalProjectsStr != null && !additionalProjectsStr.trim().isEmpty()) {
      List<String> additionalProjects =
          Splitter.on(',').trimResults().omitEmptyStrings().splitToList(additionalProjectsStr);
      for (String project : additionalProjects) {
        if (project != null && !project.isEmpty()) {
          accessibleCatalogs.add(project);
        }
      }
    }

    if (this.connection.isEnableProjectDiscovery()) {
      accessibleCatalogs.addAll(this.connection.getDiscoveredProjects());
    }

    List<String> sortedCatalogs = new ArrayList<>(accessibleCatalogs);
    Collections.sort(sortedCatalogs);
    return sortedCatalogs;
  }

  private boolean equalsOrNullMatchesAll(String expected, String actual) {
    return expected == null || expected.equals(actual);
  }

  private List<DatasetId> getTargetDatasets(String catalog, String schemaPattern, boolean isPattern)
      throws SQLException {
    if (schemaPattern != null && (!isPattern || !hasWildcards(schemaPattern))) {
      List<String> projects = resolveTargetProjects(catalog);
      List<DatasetId> datasetIds = new ArrayList<>(projects.size());
      for (String project : projects) {
        datasetIds.add(DatasetId.of(project, schemaPattern));
      }
      return datasetIds;
    }

    Pattern schemaRegex = compileSqlLikePattern(schemaPattern);
    List<Dataset> datasets = fetchMatchingDatasets(catalog, schemaPattern, schemaRegex);
    List<DatasetId> datasetIds = new ArrayList<>(datasets.size());
    for (Dataset dataset : datasets) {
      datasetIds.add(dataset.getDatasetId());
    }
    return datasetIds;
  }

  private List<DatasetId> getTargetDatasets(String catalog, String schemaPattern)
      throws SQLException {
    return getTargetDatasets(catalog, schemaPattern, true);
  }

  private List<String> resolveTargetProjects(String catalog) throws SQLException {
    return (catalog != null) ? Collections.singletonList(catalog) : getAccessibleCatalogNames();
  }

  @FunctionalInterface
  private interface TableProcessor {
    void process(Table bqTable, List<FieldValueList> collectedResults, FieldList resultSchemaFields)
        throws SQLException;
  }

  private void processSingleTable(
      DatasetId datasetId,
      String tableName,
      Table preFetchedTable,
      boolean requireBaseTable,
      List<FieldValueList> collectedResults,
      FieldList resultSchemaFields,
      TableProcessor processor,
      TableOption... options)
      throws SQLException {
    Table bqTable = preFetchedTable;
    if (bqTable == null) {
      try {
        bqTable =
            bigquery.getTable(
                TableId.of(datasetId.getProject(), datasetId.getDataset(), tableName), options);
      } catch (BigQueryException e) {
        if (e.getCode() == 404) {
          LOG.info(
              "Table '%s' or dataset '%s' not found in project '%s' (API error 404). Skipping.",
              tableName, datasetId.getDataset(), datasetId.getProject());
          return;
        }
        throw new SQLException("Error while fetching table metadata: " + e.getMessage(), e);
      } catch (Exception e) {
        throw new SQLException("Error while fetching table metadata: " + e.getMessage(), e);
      }
    }

    if (bqTable == null || bqTable.getDefinition() == null) {
      return;
    }
    if (requireBaseTable && bqTable.getDefinition().getType() != TableDefinition.Type.TABLE) {
      return;
    }
    processor.process(bqTable, collectedResults, resultSchemaFields);
  }

  private interface RoutineProcessor {
    void process(
        Routine routine, List<FieldValueList> collectedResults, FieldList resultSchemaFields)
        throws SQLException;
  }

  private void processSingleRoutine(
      DatasetId datasetId,
      String routineName,
      Routine preFetchedRoutine,
      List<FieldValueList> collectedResults,
      FieldList resultSchemaFields,
      RoutineProcessor processor,
      RoutineOption... options)
      throws SQLException {
    Routine routine = preFetchedRoutine;
    if (routine == null) {
      try {
        routine =
            bigquery.getRoutine(
                RoutineId.of(datasetId.getProject(), datasetId.getDataset(), routineName), options);
      } catch (BigQueryException e) {
        if (e.getCode() == 404) {
          LOG.info("Routine '%s' not found. Skipping.", routineName);
          return;
        }
        throw new SQLException("Error while fetching routine metadata: " + e.getMessage(), e);
      } catch (Exception e) {
        throw new SQLException("Error while fetching routine metadata: " + e.getMessage(), e);
      }
    }

    if (routine == null) {
      return;
    }
    processor.process(routine, collectedResults, resultSchemaFields);
  }

  private void processTargetRoutinesConcurrently(
      List<DatasetId> targetDatasets,
      String routineNamePattern,
      boolean isPattern,
      boolean requiresFullRoutine,
      List<FieldValueList> collectedResults,
      FieldList resultSchemaFields,
      RoutineProcessor processor,
      RoutineOption... options)
      throws SQLException {

    boolean hasWildcards = isPattern && hasWildcards(routineNamePattern);
    Pattern routineRegex = compileSqlLikePattern(routineNamePattern);
    boolean isLiteralName = routineNamePattern != null && !hasWildcards;

    ExecutorService apiExecutor = connection.getMetadataExecutor();
    List<Future<?>> activeFutures = new ArrayList<>();

    try {
      List<Callable<Void>> detailTasks = new ArrayList<>();

      if (isLiteralName) {
        for (DatasetId datasetId : targetDatasets) {
          detailTasks.add(
              () -> {
                processSingleRoutine(
                    datasetId,
                    routineNamePattern,
                    null,
                    collectedResults,
                    resultSchemaFields,
                    processor,
                    options);
                return null;
              });
        }
      } else {
        List<Future<List<Callable<Void>>>> listFutures = new ArrayList<>();
        for (DatasetId datasetId : targetDatasets) {
          Future<List<Callable<Void>>> future =
              apiExecutor.submit(
                  Context.current()
                      .wrap(
                          () -> {
                            List<Callable<Void>> tasks = new ArrayList<>();
                            try {
                              Page<Routine> routinesPage =
                                  bigquery.listRoutines(
                                      datasetId,
                                      BigQuery.RoutineListOption.pageSize(DEFAULT_PAGE_SIZE));
                              for (Routine routine : routinesPage.iterateAll()) {
                                if (Thread.currentThread().isInterrupted()) {
                                  throw new InterruptedException(
                                      "Interrupted while listing routines.");
                                }
                                String routineName = routine.getRoutineId().getRoutine();
                                if (routineRegex != null
                                    && !routineRegex.matcher(routineName).matches()) {
                                  continue;
                                }
                                tasks.add(
                                    () -> {
                                      processSingleRoutine(
                                          datasetId,
                                          requiresFullRoutine ? routineName : null,
                                          requiresFullRoutine ? null : routine,
                                          collectedResults,
                                          resultSchemaFields,
                                          processor,
                                          options);
                                      return null;
                                    });
                              }
                            } catch (BigQueryException e) {
                              if (e.getCode() == 404) {
                                LOG.info(
                                    "Dataset '%s' not found while listing routines. Skipping.",
                                    datasetId);
                              } else {
                                throw new SQLException(
                                    "Error while listing routines: " + e.getMessage(), e);
                              }
                            }
                            return tasks;
                          }));
          listFutures.add(future);
          activeFutures.add(future);
        }

        for (Future<List<Callable<Void>>> future : listFutures) {
          detailTasks.addAll(future.get());
        }
        activeFutures.clear(); // Clear so we don't hold references unnecessarily
      }

      for (Callable<Void> task : detailTasks) {
        activeFutures.add(apiExecutor.submit(Context.current().wrap(task)));
      }
      waitForTasksCompletion(activeFutures);
    } catch (ExecutionException e) {
      Throwable cause = e.getCause();
      if (cause instanceof SQLException) {
        throw (SQLException) cause;
      }
      throw new SQLException("Error while fetching metadata", e);
    } catch (InterruptedException e) {
      Thread.currentThread().interrupt();
      throw new BigQueryJdbcException("Interrupted while processing routines.", e);
    } finally {
      activeFutures.forEach(future -> future.cancel(true));
    }
  }

  private void processTargetTablesConcurrently(
      List<DatasetId> targetDatasets,
      String tableNamePattern,
      boolean isPattern,
      boolean requireBaseTable,
      boolean requiresFullTable,
      List<FieldValueList> collectedResults,
      FieldList resultSchemaFields,
      TableProcessor processor,
      TableOption... options)
      throws SQLException {

    boolean hasWildcards = isPattern && hasWildcards(tableNamePattern);
    Pattern tableRegex = compileSqlLikePattern(tableNamePattern);
    boolean isLiteralName = tableNamePattern != null && !hasWildcards;

    ExecutorService apiExecutor = connection.getMetadataExecutor();
    List<Future<?>> activeFutures = new ArrayList<>();

    try {
      // Stage 1: Discover target tables across datasets.
      List<Callable<Void>> detailTasks = new ArrayList<>();

      // For literal names, bypass listTables to avoid unnecessary API calls
      if (isLiteralName) {
        for (DatasetId datasetId : targetDatasets) {
          detailTasks.add(
              () -> {
                processSingleTable(
                    datasetId,
                    tableNamePattern,
                    null,
                    requireBaseTable,
                    collectedResults,
                    resultSchemaFields,
                    processor,
                    options);
                return null;
              });
        }
        // For wildcards/nulls, list matching tables across target datasets.
      } else {
        List<Future<List<Callable<Void>>>> listFutures = new ArrayList<>();
        for (DatasetId datasetId : targetDatasets) {
          Future<List<Callable<Void>>> future =
              apiExecutor.submit(
                  Context.current()
                      .wrap(
                          () -> {
                            List<Callable<Void>> tasks = new ArrayList<>();
                            try {
                              Page<Table> tablesPage =
                                  bigquery.listTables(
                                      datasetId,
                                      BigQuery.TableListOption.pageSize(DEFAULT_PAGE_SIZE));
                              for (Table table : tablesPage.iterateAll()) {
                                if (Thread.currentThread().isInterrupted()) {
                                  throw new InterruptedException(
                                      "Interrupted while listing tables.");
                                }
                                if (requireBaseTable
                                    && table.getDefinition() != null
                                    && table.getDefinition().getType()
                                        != TableDefinition.Type.TABLE) {
                                  continue;
                                }
                                String tableName = table.getTableId().getTable();
                                if (tableRegex != null
                                    && !tableRegex.matcher(tableName).matches()) {
                                  continue;
                                }
                                // Collect task in memory to be executed in Stage 2 after listing
                                // completes.
                                tasks.add(
                                    () -> {
                                      processSingleTable(
                                          datasetId,
                                          requiresFullTable ? tableName : null,
                                          requiresFullTable ? null : table,
                                          requireBaseTable,
                                          collectedResults,
                                          resultSchemaFields,
                                          processor,
                                          options);
                                      return null;
                                    });
                              }
                            } catch (BigQueryException e) {
                              if (e.getCode() == 404) {
                                LOG.info(
                                    "Dataset '%s' not found while listing tables. Skipping.",
                                    datasetId);
                              } else {
                                throw new SQLException(
                                    "Error while listing tables: " + e.getMessage(), e);
                              }
                            }
                            return tasks;
                          }));
          listFutures.add(future);
          activeFutures.add(future);
        }

        // Wait for listing tasks to finish before submitting detail tasks to the pool.
        for (Future<List<Callable<Void>>> future : listFutures) {
          detailTasks.addAll(future.get());
        }
        activeFutures.clear(); // Clear so we don't hold references unnecessarily
      }

      // Stage 2: Execute table metadata processing concurrently.
      for (Callable<Void> task : detailTasks) {
        activeFutures.add(apiExecutor.submit(Context.current().wrap(task)));
      }
      waitForTasksCompletion(activeFutures);
    } catch (ExecutionException e) {
      Throwable cause = e.getCause();
      if (cause instanceof SQLException) {
        throw (SQLException) cause;
      }
      throw new SQLException("Error while fetching metadata", e);
    } catch (InterruptedException e) {
      Thread.currentThread().interrupt();
      throw new BigQueryJdbcException("Interrupted while processing tables.", e);
    } finally {
      activeFutures.forEach(future -> future.cancel(true));
    }
  }

  private Schema defineForeignKeyResultSetSchema() {
    return Schema.of(
        Field.of("PKTABLE_CAT", StandardSQLTypeName.STRING),
        Field.of("PKTABLE_SCHEM", StandardSQLTypeName.STRING),
        Field.of("PKTABLE_NAME", StandardSQLTypeName.STRING),
        Field.of("PKCOLUMN_NAME", StandardSQLTypeName.STRING),
        Field.of("FKTABLE_CAT", StandardSQLTypeName.STRING),
        Field.of("FKTABLE_SCHEM", StandardSQLTypeName.STRING),
        Field.of("FKTABLE_NAME", StandardSQLTypeName.STRING),
        Field.of("FKCOLUMN_NAME", StandardSQLTypeName.STRING),
        Field.of("KEY_SEQ", StandardSQLTypeName.INT64),
        Field.of("UPDATE_RULE", StandardSQLTypeName.INT64),
        Field.of("DELETE_RULE", StandardSQLTypeName.INT64),
        Field.of("FK_NAME", StandardSQLTypeName.STRING),
        Field.of("PK_NAME", StandardSQLTypeName.STRING),
        Field.of("DEFERRABILITY", StandardSQLTypeName.INT64));
  }

  private void processForeignKey(
      ForeignKey fk,
      TableId pkTableId,
      TableId fkTableId,
      List<FieldValueList> collectedResults,
      FieldList resultSchemaFields) {
    if (pkTableId == null) {
      LOG.warning(
          String.format(
              "Skipping foreign key '%s' on table '%s' because its referenced table ID is null.",
              fk.getName() != null ? fk.getName() : "unnamed", fkTableId.getTable()));
      return;
    }
    List<ColumnReference> colRefs = fk.getColumnReferences();
    if (colRefs != null) {
      for (int i = 0; i < colRefs.size(); i++) {
        ColumnReference colRef = colRefs.get(i);
        List<FieldValue> row = new ArrayList<>();
        row.add(createStringFieldValue(pkTableId.getProject())); // PKTABLE_CAT
        row.add(createStringFieldValue(pkTableId.getDataset())); // PKTABLE_SCHEM
        row.add(createStringFieldValue(pkTableId.getTable())); // PKTABLE_NAME
        row.add(createStringFieldValue(colRef.getReferencedColumn())); // PKCOLUMN_NAME
        row.add(createStringFieldValue(fkTableId.getProject())); // FKTABLE_CAT
        row.add(createStringFieldValue(fkTableId.getDataset())); // FKTABLE_SCHEM
        row.add(createStringFieldValue(fkTableId.getTable())); // FKTABLE_NAME
        row.add(createStringFieldValue(colRef.getReferencingColumn())); // FKCOLUMN_NAME
        row.add(createLongFieldValue((long) (i + 1))); // KEY_SEQ
        row.add(createNullFieldValue()); // UPDATE_RULE
        row.add(createNullFieldValue()); // DELETE_RULE
        row.add(createStringFieldValue(fk.getName())); // FK_NAME
        row.add(createNullFieldValue()); // PK_NAME
        row.add(createNullFieldValue()); // DEFERRABILITY

        collectedResults.add(FieldValueList.of(row, resultSchemaFields));
      }
    }
  }

  private Comparator<FieldValueList> definePkTableSortComparator(FieldList resultSchemaFields) {
    final int PKTABLE_CAT_IDX = resultSchemaFields.getIndex("PKTABLE_CAT");
    final int PKTABLE_SCHEM_IDX = resultSchemaFields.getIndex("PKTABLE_SCHEM");
    final int PKTABLE_NAME_IDX = resultSchemaFields.getIndex("PKTABLE_NAME");
    final int KEY_SEQ_IDX = resultSchemaFields.getIndex("KEY_SEQ");
    return Comparator.comparing(
            (FieldValueList fvl) -> getStringValueOrNull(fvl, PKTABLE_CAT_IDX),
            Comparator.nullsFirst(String::compareTo))
        .thenComparing(
            (FieldValueList fvl) -> getStringValueOrNull(fvl, PKTABLE_SCHEM_IDX),
            Comparator.nullsFirst(String::compareTo))
        .thenComparing(
            (FieldValueList fvl) -> getStringValueOrNull(fvl, PKTABLE_NAME_IDX),
            Comparator.nullsFirst(String::compareTo))
        .thenComparing(
            (FieldValueList fvl) -> getLongValueOrNull(fvl, KEY_SEQ_IDX),
            Comparator.nullsFirst(Long::compareTo));
  }

  private Comparator<FieldValueList> defineFkTableSortComparator(FieldList resultSchemaFields) {
    final int FKTABLE_CAT_IDX = resultSchemaFields.getIndex("FKTABLE_CAT");
    final int FKTABLE_SCHEM_IDX = resultSchemaFields.getIndex("FKTABLE_SCHEM");
    final int FKTABLE_NAME_IDX = resultSchemaFields.getIndex("FKTABLE_NAME");
    final int KEY_SEQ_IDX = resultSchemaFields.getIndex("KEY_SEQ");
    return Comparator.comparing(
            (FieldValueList fvl) -> getStringValueOrNull(fvl, FKTABLE_CAT_IDX),
            Comparator.nullsFirst(String::compareTo))
        .thenComparing(
            (FieldValueList fvl) -> getStringValueOrNull(fvl, FKTABLE_SCHEM_IDX),
            Comparator.nullsFirst(String::compareTo))
        .thenComparing(
            (FieldValueList fvl) -> getStringValueOrNull(fvl, FKTABLE_NAME_IDX),
            Comparator.nullsFirst(String::compareTo))
        .thenComparing(
            (FieldValueList fvl) -> getLongValueOrNull(fvl, KEY_SEQ_IDX),
            Comparator.nullsFirst(Long::compareTo));
  }

  private static synchronized String getExportedKeysSqlContent() {
    if (exportedKeysSqlContent == null) {
      exportedKeysSqlContent = readSqlFromFile(GET_EXPORTED_KEYS_SQL);
    }
    return exportedKeysSqlContent;
  }

  static synchronized String getGraphTablesDatasetSqlContent() {
    if (graphTablesDatasetSqlContent == null) {
      graphTablesDatasetSqlContent = readSqlFromFile(GET_GRAPH_TABLES_DATASET_SQL);
    }
    return graphTablesDatasetSqlContent;
  }

  static synchronized String getGraphTablesRegionSqlContent() {
    if (graphTablesRegionSqlContent == null) {
      graphTablesRegionSqlContent = readSqlFromFile(GET_GRAPH_TABLES_REGION_SQL);
    }
    return graphTablesRegionSqlContent;
  }

  static String readSqlFromFile(String filename) {
    try (InputStream in = BigQueryDatabaseMetaData.class.getResourceAsStream(filename)) {
      if (in == null) {
        throw new IllegalArgumentException("SQL file not found: " + filename);
      }
      ByteArrayOutputStream result = new ByteArrayOutputStream();
      byte[] buffer = new byte[1024];
      int length;
      while ((length = in.read(buffer)) != -1) {
        result.write(buffer, 0, length);
      }
      return result.toString(StandardCharsets.UTF_8.name());
    } catch (IOException e) {
      throw new RuntimeException("Failed to read SQL file: " + filename, e);
    }
  }

  String replaceSqlParameters(String sql, String... params) throws SQLException {
    return String.format(sql, (Object[]) params);
  }

  private void closeStatementIgnoreException(Statement stmt) {
    if (stmt == null) {
      return;
    }
    try {
      stmt.close();
    } catch (SQLException e) {
      // ignore
    }
  }

  private interface TracedMetadataOperation<T> {
    T run() throws SQLException;
  }

  private <T> T withTracing(String spanName, TracedMetadataOperation<T> operation)
      throws SQLException {
    return BigQueryJdbcOpenTelemetry.withTracing(
        spanName, this.connection, null, () -> operation.run());
  }
}