class Google::Cloud::Bigquery::Dataset

def self.from_gapi gapi, conn, access_policy_version: nil, dataset_view: nil
  new.tap do |f|
    f.gapi = gapi
    f.service = conn
    f.access_policy_version = access_policy_version
    f.dataset_view = dataset_view
  end
end

def self.new_reference project_id, dataset_id, service
  raise ArgumentError, "dataset_id is required" unless dataset_id
  new.tap do |b|
    reference_gapi = Google::Apis::BigqueryV2::DatasetReference.new \
      project_id: project_id, dataset_id: dataset_id
    b.service = service
    b.instance_variable_set :@reference, reference_gapi
  end
end

def access
  ensure_full_data!
  reload! unless resource_full?
  access_builder = Access.from_gapi @gapi
  if block_given?
    yield access_builder
    if access_builder.changed?
      @gapi.update! access: access_builder.to_gapi
      patch_gapi! :access
    end
  end
  access_builder.freeze
end

def api_url
  return nil if reference?
  ensure_full_data!
  @gapi.self_link
end

def build_access_entry target_types: nil
  params = {
    dataset: dataset_ref,
    target_types: target_types
  }.compact
  Google::Apis::BigqueryV2::DatasetAccessEntry.new(**params)
end

def create_materialized_view table_id,
                             query,
                             name: nil,
                             description: nil,
                             enable_refresh: nil,
                             refresh_interval_ms: nil
  new_view_opts = {
    table_reference:   Google::Apis::BigqueryV2::TableReference.new(
      project_id: project_id,
      dataset_id: dataset_id,
      table_id:   table_id
    ),
    friendly_name:     name,
    description:       description,
    materialized_view: Google::Apis::BigqueryV2::MaterializedViewDefinition.new(
      enable_refresh:      enable_refresh,
      query:               query,
      refresh_interval_ms: refresh_interval_ms
    )
  }.compact
  new_view = Google::Apis::BigqueryV2::Table.new(**new_view_opts)
  gapi = service.insert_table dataset_id, new_view
  Table.from_gapi gapi, service
end

def create_routine routine_id
  ensure_service!
  new_tb = Google::Apis::BigqueryV2::Routine.new(
    routine_reference: Google::Apis::BigqueryV2::RoutineReference.new(
      project_id: project_id, dataset_id: dataset_id, routine_id: routine_id
    )
  )
  updater = Routine::Updater.new new_tb
  yield updater if block_given?
  gapi = service.insert_routine dataset_id, updater.to_gapi
  Routine.from_gapi gapi, service
end

def create_table table_id, name: nil, description: nil
  ensure_service!
  new_tb = Google::Apis::BigqueryV2::Table.new(
    table_reference: Google::Apis::BigqueryV2::TableReference.new(
      project_id: project_id, dataset_id: dataset_id,
      table_id: table_id
    )
  )
  updater = Table::Updater.new(new_tb).tap do |tb|
    tb.name = name unless name.nil?
    tb.description = description unless description.nil?
  end
  yield updater if block_given?
  gapi = service.insert_table dataset_id, updater.to_gapi
  Table.from_gapi gapi, service
end

def create_view table_id,
                query,
                name: nil,
                description: nil,
                standard_sql: nil,
                legacy_sql: nil,
                udfs: nil
  use_legacy_sql = Convert.resolve_legacy_sql standard_sql, legacy_sql
  new_view_opts = {
    table_reference: Google::Apis::BigqueryV2::TableReference.new(
      project_id: project_id,
      dataset_id: dataset_id,
      table_id:   table_id
    ),
    friendly_name:   name,
    description:     description,
    view:            Google::Apis::BigqueryV2::ViewDefinition.new(
      query:                           query,
      use_legacy_sql:                  use_legacy_sql,
      user_defined_function_resources: udfs_gapi(udfs)
    )
  }.compact
  new_view = Google::Apis::BigqueryV2::Table.new(**new_view_opts)
  gapi = service.insert_table dataset_id, new_view
  Table.from_gapi gapi, service
end

def created_at
  return nil if reference?
  ensure_full_data!
  Convert.millis_to_time @gapi.creation_time
end

def dataset_id
  return reference.dataset_id if reference?
  @gapi.dataset_reference.dataset_id
end

def dataset_ref
  dataset_ref = reference? ? reference : @gapi.dataset_reference
  dataset_ref = dataset_ref.to_h if dataset_ref.respond_to? :to_h
  dataset_ref
end

def default_collation
  return nil if reference?
  ensure_full_data!
  @gapi.default_collation
end

def default_collation= new_default_collation
  reload! unless resource_full?
  @gapi.update! default_collation: new_default_collation
  patch_gapi! :default_collation
end

def default_encryption
  return nil if reference?
  ensure_full_data!
  return nil if @gapi.default_encryption_configuration.nil?
  EncryptionConfiguration.from_gapi(@gapi.default_encryption_configuration).freeze
end

def default_encryption= value
  ensure_full_data!
  @gapi.default_encryption_configuration = value.to_gapi
  patch_gapi! :default_encryption_configuration
end

def default_expiration
  return nil if reference?
  ensure_full_data!
  begin
    Integer @gapi.default_table_expiration_ms
  rescue StandardError
    nil
  end
end

def default_expiration= new_default_expiration
  reload! unless resource_full?
  @gapi.update! default_table_expiration_ms: new_default_expiration
  patch_gapi! :default_table_expiration_ms
end

def delete force: nil
  ensure_service!
  service.delete_dataset dataset_id, force
  # Set flag for #exists?
  @exists = false
  true
end

def description
  return nil if reference?
  ensure_full_data!
  @gapi.description
end

def description= new_description
  reload! unless resource_full?
  @gapi.update! description: new_description
  patch_gapi! :description
end

def ensure_full_data!
  reload! unless resource_full?
end

def ensure_gapi!
  ensure_service!
  return unless reference?
  reload!
end

def ensure_job_succeeded! job
  return unless job.failed?
  begin
    # raise to activate ruby exception cause handling
    raise job.gapi_error
  rescue StandardError => e
    # wrap Google::Apis::Error with Google::Cloud::Error
    raise Google::Cloud::Error.from_error(e)
  end
end

def ensure_service!
  raise "Must have active connection" unless service
end

def etag
  return nil if reference?
  ensure_full_data!
  @gapi.etag
end

def exists? force: false
  return gapi_exists? if force
  # If we have a memoized value, return it
  return @exists unless @exists.nil?
  # Always true if we have a gapi object
  return true if resource?
  gapi_exists?
end

def external url, format: nil
  ext = External.from_urls url, format
  yield ext if block_given?
  ext
end

def gapi_exists?
  reload!
  @exists = true
rescue Google::Cloud::NotFoundError
  @exists = false
end

def initialize
  @service = nil
  @gapi = nil
  @reference = nil
  @access_policy_version = nil
  @dataset_view = nil
end

def insert table_id, rows, insert_ids: nil, skip_invalid: nil, ignore_unknown: nil, autocreate: nil, &block
  rows = [rows] if rows.is_a? Hash
  raise ArgumentError, "No rows provided" if rows.empty?
  insert_ids = Array.new(rows.count) { :skip } if insert_ids == :skip
  insert_ids = Array insert_ids
  if insert_ids.count.positive? && insert_ids.count != rows.count
    raise ArgumentError, "insert_ids must be the same size as rows"
  end
  if autocreate
    insert_data_with_autocreate table_id, rows, skip_invalid: skip_invalid, ignore_unknown: ignore_unknown,
                                                insert_ids: insert_ids, &block
  else
    insert_data table_id, rows, skip_invalid: skip_invalid, ignore_unknown: ignore_unknown,
                                insert_ids: insert_ids
  end
end

def insert_async table_id, skip_invalid: nil, ignore_unknown: nil, max_bytes: 10_000_000, max_rows: 500,
                 interval: 10, threads: 4, view: nil, &block
  ensure_service!
  # Get table, don't use Dataset#table which handles NotFoundError
  gapi = service.get_project_table project_id, dataset_id, table_id, metadata_view: view
  table = Table.from_gapi gapi, service, metadata_view: view
  # Get the AsyncInserter from the table
  table.insert_async skip_invalid: skip_invalid,
                     ignore_unknown: ignore_unknown,
                     max_bytes: max_bytes, max_rows: max_rows,
                     interval: interval, threads: threads, &block
end

def insert_data table_id, rows, skip_invalid: nil, ignore_unknown: nil, insert_ids: nil
  rows = [rows] if rows.is_a? Hash
  raise ArgumentError, "No rows provided" if rows.empty?
  ensure_service!
  gapi = service.insert_tabledata dataset_id, table_id, rows, skip_invalid:   skip_invalid,
                                                              ignore_unknown: ignore_unknown,
                                                              insert_ids:     insert_ids,
                                                              project_id:     project_id
  InsertResponse.from_gapi rows, gapi
end

def insert_data_with_autocreate table_id, rows, skip_invalid: nil, ignore_unknown: nil, insert_ids: nil
  insert_data table_id, rows, skip_invalid: skip_invalid, ignore_unknown: ignore_unknown, insert_ids: insert_ids
rescue Google::Cloud::NotFoundError
  sleep rand(1..60)
  begin
    create_table table_id do |tbl_updater|
      yield tbl_updater if block_given?
    end
  rescue Google::Cloud::AlreadyExistsError
    # Do nothing if it already exists
  end
  sleep 60
  retry
end

def labels
  return nil if reference?
  m = @gapi.labels
  m = m.to_h if m.respond_to? :to_h
  m.dup.freeze
end

def labels= labels
  reload! unless resource_full?
  @gapi.labels = labels
  patch_gapi! :labels
end

def load table_id, files, format: nil, create: nil, write: nil, projection_fields: nil, jagged_rows: nil,
         quoted_newlines: nil, encoding: nil, delimiter: nil, ignore_unknown: nil, max_bad_records: nil,
         quote: nil, skip_leading: nil, schema: nil, autodetect: nil, null_marker: nil, session_id: nil,
         date_format: nil, datetime_format: nil, time_format: nil, timestamp_format: nil,
         null_markers: nil, source_column_match: nil, time_zone: nil, reference_file_schema_uri: nil,
         preserve_ascii_control_characters: nil,
         reservation: nil, &block
  job = load_job table_id, files,
                 format: format, create: create, write: write, projection_fields: projection_fields,
                 jagged_rows: jagged_rows, quoted_newlines: quoted_newlines, encoding: encoding,
                 delimiter: delimiter, ignore_unknown: ignore_unknown, max_bad_records: max_bad_records,
                 quote: quote, skip_leading: skip_leading, schema: schema, autodetect: autodetect,
                 null_marker: null_marker, session_id: session_id, date_format: date_format,
                 datetime_format: datetime_format, time_format: time_format, timestamp_format: timestamp_format,
                 null_markers: null_markers, source_column_match: source_column_match, time_zone: time_zone,
                 reference_file_schema_uri: reference_file_schema_uri,
                 preserve_ascii_control_characters: preserve_ascii_control_characters,
                 reservation: reservation, &block
  job.wait_until_done!
  ensure_job_succeeded! job
  true
end

def load_job table_id, files, format: nil, create: nil, write: nil, projection_fields: nil, jagged_rows: nil,
             quoted_newlines: nil, encoding: nil, delimiter: nil, ignore_unknown: nil, max_bad_records: nil,
             quote: nil, skip_leading: nil, schema: nil, job_id: nil, prefix: nil, labels: nil, autodetect: nil,
             null_marker: nil, dryrun: nil, create_session: nil, session_id: nil, date_format: nil,
             datetime_format: nil, time_format: nil, timestamp_format: nil, null_markers: nil,
             source_column_match: nil, time_zone: nil, reference_file_schema_uri: nil,
             preserve_ascii_control_characters: nil, reservation: nil
  ensure_service!
  updater = load_job_updater table_id,
                             format: format, create: create, write: write, projection_fields: projection_fields,
                             jagged_rows: jagged_rows, quoted_newlines: quoted_newlines, encoding: encoding,
                             delimiter: delimiter, ignore_unknown: ignore_unknown,
                             max_bad_records: max_bad_records, quote: quote, skip_leading: skip_leading,
                             dryrun: dryrun, schema: schema, job_id: job_id, prefix: prefix, labels: labels,
                             autodetect: autodetect, null_marker: null_marker, create_session: create_session,
                             session_id: session_id, date_format: date_format, datetime_format: datetime_format,
                             time_format: time_format, timestamp_format: timestamp_format,
                             null_markers: null_markers, source_column_match: source_column_match,
                             time_zone: time_zone, reference_file_schema_uri: reference_file_schema_uri,
                             preserve_ascii_control_characters: preserve_ascii_control_characters,
                             reservation: reservation
  yield updater if block_given?
  load_local_or_uri files, updater
end

def load_job_csv_options! job, jagged_rows: nil, quoted_newlines: nil, delimiter: nil, quote: nil,
                          skip_leading: nil, null_marker: nil, null_markers: nil, source_column_match: nil,
                          preserve_ascii_control_characters: nil
  job.jagged_rows = jagged_rows unless jagged_rows.nil?
  job.quoted_newlines = quoted_newlines unless quoted_newlines.nil?
  job.delimiter = delimiter unless delimiter.nil?
  job.null_marker = null_marker unless null_marker.nil?
  job.quote = quote unless quote.nil?
  job.skip_leading = skip_leading unless skip_leading.nil?
  job.null_markers = null_markers unless null_markers.nil?
  job.source_column_match = source_column_match unless source_column_match.nil?
  job.preserve_ascii_control_characters = preserve_ascii_control_characters unless
    preserve_ascii_control_characters.nil?
end

def load_job_file_options! job, format: nil, projection_fields: nil, jagged_rows: nil, quoted_newlines: nil,
                           encoding: nil, delimiter: nil, ignore_unknown: nil, max_bad_records: nil, quote: nil,
                           skip_leading: nil, null_marker: nil, date_format: nil, datetime_format: nil,
                           time_format: nil, timestamp_format: nil, null_markers: nil, source_column_match: nil,
                           time_zone: nil, reference_file_schema_uri: nil,
                           preserve_ascii_control_characters: nil
  job.format = format unless format.nil?
  job.projection_fields = projection_fields unless projection_fields.nil?
  job.encoding = encoding unless encoding.nil?
  job.ignore_unknown = ignore_unknown unless ignore_unknown.nil?
  job.max_bad_records = max_bad_records unless max_bad_records.nil?
  job.date_format = date_format unless date_format.nil?
  job.datetime_format = datetime_format unless datetime_format.nil?
  job.time_format = time_format unless time_format.nil?
  job.timestamp_format = timestamp_format unless timestamp_format.nil?
  job.time_zone = time_zone unless time_zone.nil?
  job.reference_file_schema_uri = reference_file_schema_uri unless reference_file_schema_uri.nil?
  load_job_csv_options! job, jagged_rows:     jagged_rows,
                             quoted_newlines: quoted_newlines,
                             delimiter:       delimiter,
                             quote:           quote,
                             skip_leading:    skip_leading,
                             null_marker:     null_marker,
                             null_markers:    null_markers,
                             source_column_match: source_column_match,
                             preserve_ascii_control_characters: preserve_ascii_control_characters
end

def load_job_gapi table_id, dryrun, job_id: nil, prefix: nil, reservation: nil
  job_ref = service.job_ref_from job_id, prefix
  Google::Apis::BigqueryV2::Job.new(
    job_reference: job_ref,
    configuration: Google::Apis::BigqueryV2::JobConfiguration.new(
      load:    Google::Apis::BigqueryV2::JobConfigurationLoad.new(
        destination_table: Google::Apis::BigqueryV2::TableReference.new(
          project_id: project_id,
          dataset_id: dataset_id,
          table_id:   table_id
        )
      ),
      dry_run: dryrun,
      reservation: reservation
    )
  )
end

def load_job_updater table_id, format: nil, create: nil, write: nil, projection_fields: nil, jagged_rows: nil,
                     quoted_newlines: nil, encoding: nil, delimiter: nil, ignore_unknown: nil,
                     max_bad_records: nil, quote: nil, skip_leading: nil, dryrun: nil, schema: nil, job_id: nil,
                     prefix: nil, labels: nil, autodetect: nil, null_marker: nil, create_session: nil,
                     session_id: nil, date_format: nil, datetime_format: nil, time_format: nil,
                     timestamp_format: nil, null_markers: nil, source_column_match: nil, time_zone: nil,
                     reference_file_schema_uri: nil, preserve_ascii_control_characters: nil, reservation: nil
  new_job = load_job_gapi table_id, dryrun, job_id: job_id, prefix: prefix, reservation: reservation
  LoadJob::Updater.new(new_job).tap do |job|
    job.location = location if location # may be dataset reference
    job.create = create unless create.nil?
    job.write = write unless write.nil?
    job.schema = schema unless schema.nil?
    job.autodetect = autodetect unless autodetect.nil?
    job.labels = labels unless labels.nil?
    job.create_session = create_session unless create_session.nil?
    job.session_id = session_id unless session_id.nil?
    load_job_file_options! job, format:            format,
                                projection_fields: projection_fields,
                                jagged_rows:       jagged_rows,
                                quoted_newlines:   quoted_newlines,
                                encoding:          encoding,
                                delimiter:         delimiter,
                                ignore_unknown:    ignore_unknown,
                                max_bad_records:   max_bad_records,
                                quote:             quote,
                                skip_leading:      skip_leading,
                                null_marker:       null_marker,
                                date_format:       date_format,
                                datetime_format:  datetime_format,
                                time_format:       time_format,
                                timestamp_format:  timestamp_format,
                                null_markers:      null_markers,
                                source_column_match: source_column_match,
                                time_zone:         time_zone,
                                reference_file_schema_uri: reference_file_schema_uri,
                                preserve_ascii_control_characters: preserve_ascii_control_characters
  end
end

def load_local file, job_gapi
  path = Pathname(file).to_path
  if job_gapi.configuration.load.source_format.nil?
    source_format = Convert.derive_source_format path
    job_gapi.configuration.load.source_format = source_format unless source_format.nil?
  end
  gapi = service.load_table_file file, job_gapi
  Job.from_gapi gapi, service
end

def load_local_or_uri file, updater
  job_gapi = updater.to_gapi
  if local_file? file
    load_local file, job_gapi
  else
    load_storage file, job_gapi
  end
end

def load_storage urls, job_gapi
  # Convert to storage URL
  urls = [urls].flatten.map do |url|
    if url.respond_to? :to_gs_url
      url.to_gs_url
    elsif url.is_a? URI
      url.to_s
    else
      url
    end
  end
  unless urls.nil?
    job_gapi.configuration.load.update! source_uris: urls
    if job_gapi.configuration.load.source_format.nil?
      source_format = Convert.derive_source_format_from_list urls
      job_gapi.configuration.load.source_format = source_format unless source_format.nil?
    end
  end
  gapi = service.load_table_gs_url job_gapi
  Job.from_gapi gapi, service
end

def local_file? file
  ::File.file? file
rescue StandardError
  false
end

def location
  return nil if reference?
  @gapi.location
end

def model model_id, skip_lookup: nil
  ensure_service!
  return Model.new_reference project_id, dataset_id, model_id, service if skip_lookup
  gapi = service.get_model dataset_id, model_id
  Model.from_gapi_json gapi, service
rescue Google::Cloud::NotFoundError
  nil
end

def models token: nil, max: nil
  ensure_service!
  gapi = service.list_models dataset_id, token: token, max: max
  Model::List.from_gapi gapi, service, dataset_id, max
end

def modified_at
  return nil if reference?
  ensure_full_data!
  Convert.millis_to_time @gapi.last_modified_time
end

def name
  return nil if reference?
  @gapi.friendly_name
end

def name= new_name
  reload! unless resource_full?
  @gapi.update! friendly_name: new_name
  patch_gapi! :friendly_name
end

def patch_gapi! *attributes
  return if attributes.empty?
  ensure_service!
  patch_args = attributes.to_h { |attr| [attr, @gapi.send(attr)] }
  update_mode = nil
  has_access_key = patch_args.key? :access
  other_keys_exist = (patch_args.keys - [:access]).any?
  if has_access_key && !other_keys_exist
    update_mode = UpdateMode::UPDATE_ACL
  elsif !has_access_key && other_keys_exist
    update_mode = UpdateMode::UPDATE_METADATA
  elsif has_access_key && other_keys_exist
    update_mode = UpdateMode::FULL
  end
  patch_gapi = Google::Apis::BigqueryV2::Dataset.new(**patch_args)
  patch_gapi.etag = etag if etag
  @gapi = service.patch_dataset dataset_id, patch_gapi, access_policy_version: @access_policy_version,
ode: update_mode
end

def project_id
  return reference.project_id if reference?
  @gapi.dataset_reference.project_id
end

def query query,
          params: nil,
          types: nil,
          external: nil,
          max: nil,
          cache: true,
          standard_sql: nil,
          legacy_sql: nil,
          session_id: nil,
          format_options_use_int64_timestamp: true,
          reservation: nil,
          &block
  job = query_job query,
                  params: params,
                  types: types,
                  external: external,
                  cache: cache,
                  standard_sql: standard_sql,
                  legacy_sql: legacy_sql,
                  session_id: session_id,
                  reservation: reservation,
                  &block
  job.wait_until_done!
  ensure_job_succeeded! job
  job.data max: max, format_options_use_int64_timestamp: format_options_use_int64_timestamp
end

def query_job query,
              params: nil,
              types: nil,
              external: nil,
              priority: "INTERACTIVE",
              cache: true,
              table: nil,
              create: nil,
              write: nil,
              dryrun: nil,
              standard_sql: nil,
              legacy_sql: nil,
              large_results: nil,
              flatten: nil,
              maximum_billing_tier: nil,
              maximum_bytes_billed: nil,
              job_id: nil,
              prefix: nil,
              labels: nil,
              udfs: nil,
              create_session: nil,
              session_id: nil,
              reservation: nil
  ensure_service!
  options = {
    params: params,
    types: types,
    external: external,
    priority: priority,
    cache: cache,
    table: table,
    create: create,
    write: write,
    dryrun: dryrun,
    standard_sql: standard_sql,
    legacy_sql: legacy_sql,
    large_results: large_results,
    flatten: flatten,
    maximum_billing_tier: maximum_billing_tier,
    maximum_bytes_billed: maximum_bytes_billed,
    job_id: job_id,
    prefix: prefix,
    labels: labels,
    udfs: udfs,
    create_session: create_session,
    session_id: session_id,
    reservation: reservation
  }
  updater = QueryJob::Updater.from_options service, query, options
  updater.dataset = self
  updater.location = location if location # may be dataset reference
  yield updater if block_given?
  gapi = service.query_job updater.to_gapi
  Job.from_gapi gapi, service
end

def reference?
  @gapi.nil?
end

def reload!
  ensure_service!
  @gapi = service.get_project_dataset project_id, dataset_id, access_policy_version: @access_policy_version,
    dataset_view: @dataset_view
  @reference = nil
  @exists = nil
  self
end

def resource?
  !@gapi.nil?
end

def resource_full?
  @gapi.is_a? Google::Apis::BigqueryV2::Dataset
end

def resource_partial?
  @gapi.is_a? Google::Apis::BigqueryV2::DatasetList::Dataset
end

def routine routine_id, skip_lookup: nil
  ensure_service!
  return Routine.new_reference project_id, dataset_id, routine_id, service if skip_lookup
  gapi = service.get_routine dataset_id, routine_id
  Routine.from_gapi gapi, service
rescue Google::Cloud::NotFoundError
  nil
end

def routines token: nil, max: nil, filter: nil
  ensure_service!
  gapi = service.list_routines dataset_id, token: token, max: max, filter: filter
  Routine::List.from_gapi gapi, service, dataset_id, max, filter: filter
end

def storage_billing_model
  return nil if reference?
  ensure_full_data!
  @gapi.storage_billing_model
end

def storage_billing_model= value
  ensure_full_data!
  @gapi.storage_billing_model = value
  patch_gapi! :storage_billing_model
end

def storage_url? files
  [files].flatten.all? do |file|
    file.respond_to?(:to_gs_url) ||
      (file.respond_to?(:to_str) && file.to_str.downcase.start_with?("gs://")) ||
      (file.is_a?(URI) && file.to_s.downcase.start_with?("gs://"))
  end
end

def table table_id, skip_lookup: nil, view: nil
  ensure_service!
  return Table.new_reference project_id, dataset_id, table_id, service if skip_lookup
  gapi = service.get_project_table project_id, dataset_id, table_id, metadata_view: view
  Table.from_gapi gapi, service, metadata_view: view
rescue Google::Cloud::NotFoundError
  nil
end

def tables token: nil, max: nil
  ensure_service!
  gapi = service.list_tables dataset_id, token: token, max: max
  Table::List.from_gapi gapi, service, dataset_id, max
end

def tags
  ensure_full_data!
  return nil if @gapi.tags.nil?
  @gapi.tags.map { |gapi| Tag.from_gapi gapi }
end

def udfs_gapi array_or_str
  return [] if array_or_str.nil?
  Array(array_or_str).map do |uri_or_code|
    resource = Google::Apis::BigqueryV2::UserDefinedFunctionResource.new
    if uri_or_code.start_with? "gs://"
      resource.resource_uri = uri_or_code
    else
      resource.inline_code = uri_or_code
    end
    resource
  end
end