feat(bigquery): load jobs with dest tables in another project

google-cloud-bigquery/acceptance/bigquery/table_test.rb

@@ -182,6 +182,7 @@
   end
 
   it "loads and reloads table with partial projection of table metadata" do
+    _(table.table_id).must_equal table_id # ensure table is created
     %w[unspecified basic storage full].each do |view|
       table = dataset.table table_id, view: view
       _(table.table_id).must_equal table_id

google-cloud-bigquery/lib/google/cloud/bigquery/dataset.rb

@@ -899,7 +899,7 @@ def create_materialized_view table_id,
         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_table dataset_id, table_id, metadata_view: view
+          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
@@ -2444,7 +2444,7 @@ def load table_id, files, format: nil, create: nil, write: nil, projection_field
         #
         def reload!
           ensure_service!
-          @gapi = service.get_dataset dataset_id
+          @gapi = service.get_project_dataset project_id, dataset_id
           @reference = nil
           @exists = nil
           self
@@ -2927,7 +2927,7 @@ def load_job_gapi table_id, dryrun, job_id: nil, prefix: nil
             configuration: Google::Apis::BigqueryV2::JobConfiguration.new(
               load:    Google::Apis::BigqueryV2::JobConfigurationLoad.new(
                 destination_table: Google::Apis::BigqueryV2::TableReference.new(
-                  project_id: @service.project,
+                  project_id: project_id,
                   dataset_id: dataset_id,
                   table_id:   table_id
                 )

google-cloud-bigquery/lib/google/cloud/bigquery/project.rb

@@ -1111,6 +1111,7 @@ def query query,
         #   Note: This will work only for tables in _SESSION dataset
         #         else the property will be ignored by the backend.
         # @param [string] session_id Session ID in which the load job must run.
+        # @param [string] project_id Project ID where the destination table exists.
         #
         # @yield [updater] A block for setting the schema and other
         #   options for the destination table. The schema can be omitted if the
@@ -1138,10 +1139,10 @@ def load_job table_id, files, dataset_id: nil, format: nil, create: nil, write:
                      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, &block
+                     null_marker: nil, dryrun: nil, create_session: nil, session_id: nil, project_id: nil, &block
           ensure_service!
           dataset_id ||= "_SESSION" unless create_session.nil? && session_id.nil?
-          session_dataset = dataset dataset_id, skip_lookup: true
+          session_dataset = dataset dataset_id, skip_lookup: true, project_id: project_id
           table = session_dataset.table table_id, skip_lookup: true
           table.load_job  files,
                           format: format, create: create, write: write, projection_fields: projection_fields,
@@ -1376,6 +1377,7 @@ def external url, format: nil
         #   object without verifying that the resource exists on the BigQuery
         #   service. Calls made on this object will raise errors if the resource
         #   does not exist. Default is `false`. Optional.
+        # @param [String] project_id The GCP Project where the dataset lives.
         #
         # @return [Google::Cloud::Bigquery::Dataset, nil] Returns `nil` if the
         #   dataset does not exist.
@@ -1388,17 +1390,26 @@ def external url, format: nil
         #   dataset = bigquery.dataset "my_dataset"
         #   puts dataset.name
         #
+        # @example
+        #   require "google/cloud/bigquery"
+        #
+        #   bigquery = Google::Cloud::Bigquery.new
+        #
+        #   dataset = bigquery.dataset "my_dataset", project_id: "another_project"
+        #   puts dataset.name
+        #
         # @example Avoid retrieving the dataset resource with `skip_lookup`:
         #   require "google/cloud/bigquery"
         #
         #   bigquery = Google::Cloud::Bigquery.new
         #
         #   dataset = bigquery.dataset "my_dataset", skip_lookup: true
         #
-        def dataset dataset_id, skip_lookup: nil
+        def dataset dataset_id, skip_lookup: nil, project_id: nil
           ensure_service!
-          return Dataset.new_reference project, dataset_id, service if skip_lookup
-          gapi = service.get_dataset dataset_id
+          project_id ||= project
+          return Dataset.new_reference project_id, dataset_id, service if skip_lookup
+          gapi = service.get_project_dataset project_id, dataset_id
           Dataset.from_gapi gapi, service
         rescue Google::Cloud::NotFoundError
           nil

google-cloud-bigquery/lib/google/cloud/bigquery/service.rb

@@ -110,9 +110,15 @@ def list_datasets all: nil, filter: nil, max: nil, token: nil
         ##
         # Returns the dataset specified by datasetID.
         def get_dataset dataset_id
+          get_project_dataset @project, dataset_id
+        end
+
+        ##
+        # Gets the specified dataset resource by full dataset reference.
+        def get_project_dataset project_id, dataset_id
           # The get operation is considered idempotent
           execute backoff: true do
-            service.get_dataset @project, dataset_id
+            service.get_dataset project_id, dataset_id
           end
         end
 

google-cloud-bigquery/lib/google/cloud/bigquery/table.rb

@@ -3081,7 +3081,7 @@ def load_job_gapi table_id, dryrun, job_id: nil, prefix: nil
             configuration: Google::Apis::BigqueryV2::JobConfiguration.new(
               load:    Google::Apis::BigqueryV2::JobConfigurationLoad.new(
                 destination_table: Google::Apis::BigqueryV2::TableReference.new(
-                  project_id: @service.project,
+                  project_id: project_id,
                   dataset_id: dataset_id,
                   table_id:   table_id
                 )