diff --git a/google-cloud-bigquery/.rubocop.yml b/google-cloud-bigquery/.rubocop.yml
index 59711511d084..3c0cdb104a87 100644
--- a/google-cloud-bigquery/.rubocop.yml
+++ b/google-cloud-bigquery/.rubocop.yml
@@ -14,7 +14,7 @@ Style/Documentation:
 Lint/MixedRegexpCaptureTypes:
   Enabled: false
 Metrics/AbcSize:
-  Max: 50
+  Max: 55
 Metrics/BlockLength:
   Exclude:
     - "google-cloud-bigquery.gemspec"
diff --git a/google-cloud-bigquery/acceptance/bigquery/advanced_test.rb b/google-cloud-bigquery/acceptance/bigquery/advanced_test.rb
index 09fee7b040dc..88437e703016 100644
--- a/google-cloud-bigquery/acceptance/bigquery/advanced_test.rb
+++ b/google-cloud-bigquery/acceptance/bigquery/advanced_test.rb
@@ -69,6 +69,25 @@
     _(rows[2]).must_equal({ name: "Gandalf", spells_name: "Skydragon", spells_properties_name: "Explodey", spells_properties_power: 11.0 })
   end
 
+  it "queries in session mode" do
+    job = bigquery.query_job "CREATE TEMPORARY TABLE temptable AS SELECT 17 as foo", dataset: dataset, create_session: true
+    job.wait_until_done!
+    _(job).wont_be :failed?
+    _(job.session_id).wont_be :nil?
+
+    job_2 = bigquery.query_job "SELECT * FROM temptable", dataset: dataset, session_id: job.session_id
+    job_2.wait_until_done!
+    _(job_2).wont_be :failed?
+    _(job_2.session_id).wont_be :nil?
+    _(job_2.session_id).must_equal job.session_id
+    _(job_2.data.first).wont_be :nil?
+    _(job_2.data.first[:foo]).must_equal 17
+
+    data = bigquery.query "SELECT * FROM temptable", dataset: dataset, session_id: job.session_id
+    _(data.first).wont_be :nil?
+    _(data.first[:foo]).must_equal 17
+  end
+
   it "modifies a nested schema via field" do
     empty_table_id = "#{table_id}_empty"
     empty_table = dataset.table empty_table_id
diff --git a/google-cloud-bigquery/acceptance/bigquery/dataset_reference_test.rb b/google-cloud-bigquery/acceptance/bigquery/dataset_reference_test.rb
index 5f81ef6e4f46..59d43de8661b 100644
--- a/google-cloud-bigquery/acceptance/bigquery/dataset_reference_test.rb
+++ b/google-cloud-bigquery/acceptance/bigquery/dataset_reference_test.rb
@@ -218,6 +218,7 @@
     query_job = dataset.query_job query, job_id: job_id
     _(query_job).must_be_kind_of Google::Cloud::Bigquery::QueryJob
     _(query_job.job_id).must_equal job_id
+    _(query_job.session_id).must_be :nil?
     query_job.wait_until_done!
     _(query_job.done?).must_equal true
     _(query_job.data.total).wont_be_nil
diff --git a/google-cloud-bigquery/acceptance/bigquery/dataset_test.rb b/google-cloud-bigquery/acceptance/bigquery/dataset_test.rb
index dee9f1d06745..14f1080908d7 100644
--- a/google-cloud-bigquery/acceptance/bigquery/dataset_test.rb
+++ b/google-cloud-bigquery/acceptance/bigquery/dataset_test.rb
@@ -541,4 +541,23 @@
 
     assert_data table.data(max: 1)
   end
+
+  it "queries in session mode" do
+    job = dataset.query_job "CREATE TEMPORARY TABLE temptable AS SELECT 17 as foo", create_session: true
+    job.wait_until_done!
+    _(job).wont_be :failed?
+    _(job.session_id).wont_be :nil?
+
+    job_2 = dataset.query_job "SELECT * FROM temptable", session_id: job.session_id
+    job_2.wait_until_done!
+    _(job_2).wont_be :failed?
+    _(job_2.session_id).wont_be :nil?
+    _(job_2.session_id).must_equal job.session_id
+    _(job_2.data.first).wont_be :nil?
+    _(job_2.data.first[:foo]).must_equal 17
+
+    data = dataset.query "SELECT * FROM temptable", session_id: job.session_id
+    _(data.first).wont_be :nil?
+    _(data.first[:foo]).must_equal 17
+  end
 end
diff --git a/google-cloud-bigquery/lib/google/cloud/bigquery/dataset.rb b/google-cloud-bigquery/lib/google/cloud/bigquery/dataset.rb
index 6203cca37c08..33e89cfbf7c2 100644
--- a/google-cloud-bigquery/lib/google/cloud/bigquery/dataset.rb
+++ b/google-cloud-bigquery/lib/google/cloud/bigquery/dataset.rb
@@ -1244,6 +1244,8 @@ def routines token: nil, max: nil, filter: nil
         #   Flattens all nested and repeated fields in the query results. The
         #   default value is `true`. `large_results` parameter must be `true` if
         #   this is set to `false`.
+        # @param [Integer] maximum_billing_tier Deprecated: Change the billing
+        #   tier to allow high-compute queries.
         # @param [Integer] maximum_bytes_billed Limits the bytes billed for this
         #   job. Queries that will have bytes billed beyond this limit will fail
         #   (without incurring a charge). Optional. If unspecified, this will be
@@ -1294,8 +1296,12 @@ def routines token: nil, max: nil, filter: nil
         #   For additional information on migrating, see: [Migrating to
         #   standard SQL - Differences in user-defined JavaScript
         #   functions](https://cloud.google.com/bigquery/docs/reference/standard-sql/migrating-from-legacy-sql#differences_in_user-defined_javascript_functions)
-        # @param [Integer] maximum_billing_tier Deprecated: Change the billing
-        #   tier to allow high-compute queries.
+        # @param [Boolean] create_session If true, creates a new session, where the
+        #   session ID will be a server generated random id. If false, runs query
+        #   with an existing session ID when one is provided in the `session_id`
+        #   param, otherwise runs query in non-session mode. See {Job#session_id}.
+        # @param [String] session_id The ID of an existing session. See also the
+        #   `create_session` param and {Job#session_id}.
         # @yield [job] a job configuration object
         # @yieldparam [Google::Cloud::Bigquery::QueryJob::Updater] job a job
         #   configuration object for setting additional options for the query.
@@ -1435,16 +1441,52 @@ def routines token: nil, max: nil, filter: nil
         #
         # @!group Data
         #
-        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
+        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
           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 }
+          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
+          }
 
           updater = QueryJob::Updater.from_options service, query, options
           updater.dataset = self
@@ -1566,6 +1608,8 @@ def query_job query, params: nil, types: nil, external: nil, priority: "INTERACT
         #   When set to false, the values of `large_results` and `flatten` are
         #   ignored; the query will be run as if `large_results` is true and
         #   `flatten` is false. Optional. The default value is false.
+        # @param [String] session_id The ID of an existing session. See the
+        #   `create_session` param in {#query_job} and {Job#session_id}.
         # @yield [job] a job configuration object
         # @yieldparam [Google::Cloud::Bigquery::QueryJob::Updater] job a job
         #   configuration object for setting additional options for the query.
@@ -1699,10 +1743,25 @@ def query_job query, params: nil, types: nil, external: nil, priority: "INTERACT
         #
         # @!group Data
         #
-        def query query, params: nil, types: nil, external: nil, max: nil, cache: true,
-                  standard_sql: nil, legacy_sql: nil, &block
-          job = query_job query, params: params, types: types, external: external, cache: cache,
-                                 standard_sql: standard_sql, legacy_sql: legacy_sql, &block
+        def query query,
+                  params: nil,
+                  types: nil,
+                  external: nil,
+                  max: nil,
+                  cache: true,
+                  standard_sql: nil,
+                  legacy_sql: nil,
+                  session_id: 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,
+                          &block
           job.wait_until_done!
           ensure_job_succeeded! job
 
diff --git a/google-cloud-bigquery/lib/google/cloud/bigquery/job.rb b/google-cloud-bigquery/lib/google/cloud/bigquery/job.rb
index 313d582a325a..5ee1a18d514f 100644
--- a/google-cloud-bigquery/lib/google/cloud/bigquery/job.rb
+++ b/google-cloud-bigquery/lib/google/cloud/bigquery/job.rb
@@ -226,6 +226,16 @@ def reservation_usage
           Array(@gapi.statistics.reservation_usage).map { |g| ReservationUsage.from_gapi g }
         end
 
+        ##
+        # The ID of the session if this job is part of one. See the `create_session` param in {Project#query_job} and
+        # {Dataset#query_job}.
+        #
+        # @return [String, nil] The session ID, or `nil` if not associated with a session.
+        #
+        def session_id
+          @gapi.statistics.session_info&.session_id
+        end
+
         ##
         # The ID of a multi-statement transaction.
         #
diff --git a/google-cloud-bigquery/lib/google/cloud/bigquery/project.rb b/google-cloud-bigquery/lib/google/cloud/bigquery/project.rb
index adccd69bb753..53bba8a6147e 100644
--- a/google-cloud-bigquery/lib/google/cloud/bigquery/project.rb
+++ b/google-cloud-bigquery/lib/google/cloud/bigquery/project.rb
@@ -402,6 +402,8 @@ def copy source_table, destination_table, create: nil, write: nil, &block
         #   Flattens all nested and repeated fields in the query results. The
         #   default value is `true`. `large_results` parameter must be `true` if
         #   this is set to `false`.
+        # @param [Integer] maximum_billing_tier Deprecated: Change the billing
+        #   tier to allow high-compute queries.
         # @param [Integer] maximum_bytes_billed Limits the bytes billed for this
         #   job. Queries that will have bytes billed beyond this limit will fail
         #   (without incurring a charge). Optional. If unspecified, this will be
@@ -455,8 +457,12 @@ def copy source_table, destination_table, create: nil, write: nil, &block
         #   For additional information on migrating, see: [Migrating to
         #   standard SQL - Differences in user-defined JavaScript
         #   functions](https://cloud.google.com/bigquery/docs/reference/standard-sql/migrating-from-legacy-sql#differences_in_user-defined_javascript_functions)
-        # @param [Integer] maximum_billing_tier Deprecated: Change the billing
-        #   tier to allow high-compute queries.
+        # @param [Boolean] create_session If true, creates a new session, where the
+        #   session ID will be a server generated random id. If false, runs query
+        #   with an existing session ID when one is provided in the `session_id`
+        #   param, otherwise runs query in non-session mode. See {Job#session_id}.
+        # @param [String] session_id The ID of an existing session. See also the
+        #   `create_session` param and {Job#session_id}.
         # @yield [job] a job configuration object
         # @yieldparam [Google::Cloud::Bigquery::QueryJob::Updater] job a job
         #   configuration object for setting query options.
@@ -599,17 +605,56 @@ def copy source_table, destination_table, create: nil, write: nil, &block
         #     end
         #   end
         #
-        def query_job query, params: nil, types: nil, external: nil, priority: "INTERACTIVE", cache: true, table: nil,
-                      create: nil, write: nil, dryrun: nil, dataset: nil, project: 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
+        def query_job query,
+                      params: nil,
+                      types: nil,
+                      external: nil,
+                      priority: "INTERACTIVE",
+                      cache: true,
+                      table: nil,
+                      create: nil,
+                      write: nil,
+                      dryrun: nil,
+                      dataset: nil,
+                      project: 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
           ensure_service!
-          options = { params: params, types: types, external: external, priority: priority, cache: cache, table: table,
-                      create: create, write: write, dryrun: dryrun, dataset: dataset,
-                      project: (project || self.project), 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 }
+          options = {
+            params: params,
+            types: types,
+            external: external,
+            priority: priority,
+            cache: cache,
+            table: table,
+            create: create,
+            write: write,
+            dryrun: dryrun,
+            dataset: dataset,
+            project: (project || self.project),
+            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
+          }
 
           updater = QueryJob::Updater.from_options service, query, options
 
@@ -730,6 +775,8 @@ def query_job query, params: nil, types: nil, external: nil, priority: "INTERACT
         #   When set to false, the values of `large_results` and `flatten` are
         #   ignored; the query will be run as if `large_results` is true and
         #   `flatten` is false. Optional. The default value is false.
+        # @param [String] session_id The ID of an existing session. See the
+        #   `create_session` param in {#query_job} and {Job#session_id}.
         # @yield [job] a job configuration object
         # @yieldparam [Google::Cloud::Bigquery::QueryJob::Updater] job a job
         #   configuration object for setting additional options for the query.
@@ -873,10 +920,29 @@ def query_job query, params: nil, types: nil, external: nil, priority: "INTERACT
         #   # Retrieve the next page of results
         #   data = data.next if data.next?
         #
-        def query query, params: nil, types: nil, external: nil, max: nil, cache: true, dataset: nil, project: nil,
-                  standard_sql: nil, legacy_sql: nil, &block
-          job = query_job query, params: params, types: types, external: external, cache: cache, dataset: dataset,
-                                 project: project, standard_sql: standard_sql, legacy_sql: legacy_sql, &block
+        def query query,
+                  params: nil,
+                  types: nil,
+                  external: nil,
+                  max: nil,
+                  cache: true,
+                  dataset: nil,
+                  project: nil,
+                  standard_sql: nil,
+                  legacy_sql: nil,
+                  session_id: nil,
+                  &block
+          job = query_job query,
+                          params: params,
+                          types: types,
+                          external: external,
+                          cache: cache,
+                          dataset: dataset,
+                          project: project,
+                          standard_sql: standard_sql,
+                          legacy_sql: legacy_sql,
+                          session_id: session_id,
+                          &block
           job.wait_until_done!
 
           if job.failed?
diff --git a/google-cloud-bigquery/lib/google/cloud/bigquery/query_job.rb b/google-cloud-bigquery/lib/google/cloud/bigquery/query_job.rb
index 2f8adc6bfbed..03837b114fc0 100644
--- a/google-cloud-bigquery/lib/google/cloud/bigquery/query_job.rb
+++ b/google-cloud-bigquery/lib/google/cloud/bigquery/query_job.rb
@@ -775,6 +775,8 @@ def self.from_options service, query, options
             updater = QueryJob::Updater.new service, req
             updater.set_params_and_types options[:params], options[:types] if options[:params]
             updater.create = options[:create]
+            updater.create_session = options[:create_session]
+            updater.session_id = options[:session_id] if options[:session_id]
             updater.write = options[:write]
             updater.table = options[:table]
             updater.dryrun = options[:dryrun]
@@ -1018,6 +1020,37 @@ def create= value
             @gapi.configuration.query.create_disposition = Convert.create_disposition value
           end
 
+          ##
+          # Sets the create_session property. If true, creates a new session,
+          # where session id will be a server generated random id. If false,
+          # runs query with an existing {#session_id=}, otherwise runs query in
+          # non-session mode. The default value is `false`.
+          #
+          # @param [Boolean] value The create_session property. The default
+          # value is `false`.
+          #
+          # @!group Attributes
+          def create_session= value
+            @gapi.configuration.query.create_session = value
+          end
+
+          ##
+          # Sets the session ID for a query run in session mode. See {#create_session=}.
+          #
+          # @param [String] value The session ID. The default value is `nil`.
+          #
+          # @!group Attributes
+          def session_id= value
+            @gapi.configuration.query.connection_properties ||= []
+            prop = @gapi.configuration.query.connection_properties.find { |cp| cp.key == "session_id" }
+            if prop
+              prop.value = value
+            else
+              prop = Google::Apis::BigqueryV2::ConnectionProperty.new key: "session_id", value: value
+              @gapi.configuration.query.connection_properties << prop
+            end
+          end
+
           ##
           # Sets the write disposition for when the query results table exists.
           #
diff --git a/google-cloud-bigquery/test/google/cloud/bigquery/dataset_query_job_test.rb b/google-cloud-bigquery/test/google/cloud/bigquery/dataset_query_job_test.rb
index eb46e9ac9c6f..15bb58c65ce7 100644
--- a/google-cloud-bigquery/test/google/cloud/bigquery/dataset_query_job_test.rb
+++ b/google-cloud-bigquery/test/google/cloud/bigquery/dataset_query_job_test.rb
@@ -26,6 +26,7 @@
                                                   bigquery.service }
   let(:labels) { { "foo" => "bar" } }
   let(:udfs) { [ "return x+1;", "gs://my-bucket/my-lib.js" ] }
+  let(:session_id) { "mysessionid" }
   let(:range_partitioning) do
     Google::Apis::BigqueryV2::RangePartitioning.new(
       field: "my_table_id",
@@ -339,4 +340,42 @@
     _(job).must_be_kind_of Google::Cloud::Bigquery::QueryJob
     _(job.udfs).must_equal ["gs://my-bucket/my-lib.js"]
   end
+
+  it "queries the data with create_session option" do
+    mock = Minitest::Mock.new
+    bigquery.service.mocked_service = mock
+
+    job_gapi = query_job_gapi query, create_session: true
+    job_gapi.configuration.query.default_dataset = Google::Apis::BigqueryV2::DatasetReference.new(
+      project_id: project,
+      dataset_id: dataset_id
+    )
+    job_resp_gapi = query_job_resp_gapi query, session_id: session_id
+    mock.expect :insert_job, job_resp_gapi, [project, job_gapi]
+
+    job = dataset.query_job query, create_session: true
+    mock.verify
+
+    _(job).must_be_kind_of Google::Cloud::Bigquery::QueryJob
+    _(job.session_id).must_equal session_id
+  end
+
+  it "queries the data with session_id option" do
+    mock = Minitest::Mock.new
+    bigquery.service.mocked_service = mock
+
+    job_gapi = query_job_gapi query, session_id: session_id
+    job_gapi.configuration.query.default_dataset = Google::Apis::BigqueryV2::DatasetReference.new(
+      project_id: project,
+      dataset_id: dataset_id
+    )
+    job_resp_gapi = query_job_resp_gapi query, session_id: session_id
+    mock.expect :insert_job, job_resp_gapi, [project, job_gapi]
+
+    job = dataset.query_job query, session_id: session_id
+    mock.verify
+
+    _(job).must_be_kind_of Google::Cloud::Bigquery::QueryJob
+    _(job.session_id).must_equal session_id
+  end
 end
diff --git a/google-cloud-bigquery/test/google/cloud/bigquery/job_test.rb b/google-cloud-bigquery/test/google/cloud/bigquery/job_test.rb
index f8510da87d55..e138be302f22 100644
--- a/google-cloud-bigquery/test/google/cloud/bigquery/job_test.rb
+++ b/google-cloud-bigquery/test/google/cloud/bigquery/job_test.rb
@@ -20,7 +20,8 @@
   # Create a job object with the project's mocked connection object
   let(:region) { "US" }
   let(:labels) { { "foo" => "bar" } }
-  let(:job_hash) { random_job_hash location: region, transaction_id: "123456789" }
+  let(:session_id) { "mysessionid" }
+  let(:job_hash) { random_job_hash location: region, transaction_id: "123456789", session_id: session_id }
   let(:job_gapi) do
     job_gapi = Google::Apis::BigqueryV2::Job.from_json job_hash.to_json
     job_gapi.configuration.labels = labels
@@ -60,6 +61,7 @@
     _(job.user_email).must_equal "user@example.com"
     _(job.num_child_jobs).must_equal 2
     _(job.parent_job_id).must_equal "2222222222"
+    _(job.session_id).must_equal session_id
   end
 
   it "knows its state" do
diff --git a/google-cloud-bigquery/test/google/cloud/bigquery/project_query_job_test.rb b/google-cloud-bigquery/test/google/cloud/bigquery/project_query_job_test.rb
index c28bb1a4f3a6..741820474231 100644
--- a/google-cloud-bigquery/test/google/cloud/bigquery/project_query_job_test.rb
+++ b/google-cloud-bigquery/test/google/cloud/bigquery/project_query_job_test.rb
@@ -26,6 +26,7 @@
                                                   bigquery.service }
   let(:labels) { { "foo" => "bar" } }
   let(:udfs) { [ "return x+1;", "gs://my-bucket/my-lib.js" ] }
+  let(:session_id) { "mysessionid" }
 
   it "queries the data" do
     mock = Minitest::Mock.new
@@ -246,4 +247,34 @@
     _(job).must_be_kind_of Google::Cloud::Bigquery::QueryJob
     _(job.udfs).must_equal ["gs://my-bucket/my-lib.js"]
   end
+
+  it "queries the data with create_session option" do
+    mock = Minitest::Mock.new
+    bigquery.service.mocked_service = mock
+
+    job_gapi = query_job_gapi query, location: nil, create_session: true
+    job_resp_gapi = query_job_resp_gapi query, session_id: session_id
+    mock.expect :insert_job, job_resp_gapi, [project, job_gapi]
+
+    job = bigquery.query_job query, create_session: true
+    mock.verify
+
+    _(job).must_be_kind_of Google::Cloud::Bigquery::QueryJob
+    _(job.session_id).must_equal session_id
+  end
+
+  it "queries the data with session_id option" do
+    mock = Minitest::Mock.new
+    bigquery.service.mocked_service = mock
+
+    job_gapi = query_job_gapi query, location: nil, session_id: session_id
+    job_resp_gapi = query_job_resp_gapi query, session_id: session_id
+    mock.expect :insert_job, job_resp_gapi, [project, job_gapi]
+
+    job = bigquery.query_job query, session_id: session_id
+    mock.verify
+
+    _(job).must_be_kind_of Google::Cloud::Bigquery::QueryJob
+    _(job.session_id).must_equal session_id
+  end
 end
diff --git a/google-cloud-bigquery/test/helper.rb b/google-cloud-bigquery/test/helper.rb
index e07938ff6188..9f1fca2a6fbe 100644
--- a/google-cloud-bigquery/test/helper.rb
+++ b/google-cloud-bigquery/test/helper.rb
@@ -581,7 +581,7 @@ def random_routine_gapi dataset, id = nil, project_id: nil, description: nil
     Google::Apis::BigqueryV2::Routine.from_json json
   end
 
-  def random_job_hash id = "job_9876543210", state = "running", location: "US", transaction_id: nil
+  def random_job_hash id = "job_9876543210", state = "running", location: "US", transaction_id: nil, session_id: nil
     hash = {
       "kind" => "bigquery#job",
       "etag" => "etag",
@@ -628,6 +628,7 @@ def random_job_hash id = "job_9876543210", state = "running", location: "US", tr
     }
     hash["jobReference"]["location"] = location if location
     hash["statistics"]["transactionInfo"] = { "transactionId": transaction_id } if transaction_id
+    hash["statistics"]["sessionInfo"] = { "sessionId": session_id } if session_id
     hash
   end
 
@@ -677,8 +678,9 @@ def query_job_resp_gapi query,
                           ddl_operation_performed: nil,
                           deleted_row_count: nil,
                           inserted_row_count: nil,
-                          updated_row_count: nil
-    gapi = Google::Apis::BigqueryV2::Job.from_json query_job_resp_json query, job_id: job_id
+                          updated_row_count: nil,
+                          session_id: nil
+    gapi = Google::Apis::BigqueryV2::Job.from_json query_job_resp_json(query, job_id: job_id, session_id: session_id)
     gapi.statistics.query = statistics_query_gapi target_routine: target_routine,
                                                   target_table: target_table,
                                                   statement_type: statement_type,
@@ -690,7 +692,7 @@ def query_job_resp_gapi query,
     gapi
   end
 
-  def query_job_resp_json query, job_id: "job_9876543210", location: "US"
+  def query_job_resp_json query, job_id: "job_9876543210", location: "US", session_id: nil
     hash = random_job_hash job_id, "done", location: location
     hash["configuration"]["query"] = {
       "query" => query,
@@ -714,6 +716,7 @@ def query_job_resp_json query, job_id: "job_9876543210", location: "US"
       "maximumBillingTier" => nil,
       "maximumBytesBilled" => nil
     }
+    hash["statistics"]["sessionInfo"] = { "sessionId": session_id } if session_id
     hash.to_json
   end
 
@@ -812,8 +815,8 @@ def failed_query_job_resp_json query, job_id: "job_9876543210", reason: "accessD
     hash.to_json
   end
 
-  def query_job_gapi query, parameter_mode: nil, dataset: nil, job_id: "job_9876543210", location: "US", dry_run: nil
-    gapi = Google::Apis::BigqueryV2::Job.from_json query_job_json(query, job_id: job_id, location: location, dry_run: dry_run)
+  def query_job_gapi query, parameter_mode: nil, dataset: nil, job_id: "job_9876543210", location: "US", dry_run: nil, create_session: nil, session_id: nil
+    gapi = Google::Apis::BigqueryV2::Job.from_json query_job_json(query, job_id: job_id, location: location, dry_run: dry_run, create_session: create_session, session_id: session_id)
     gapi.configuration.query.parameter_mode = parameter_mode if parameter_mode
     gapi.configuration.query.default_dataset = Google::Apis::BigqueryV2::DatasetReference.new(
       dataset_id: dataset, project_id: project
@@ -821,7 +824,7 @@ def query_job_gapi query, parameter_mode: nil, dataset: nil, job_id: "job_987654
     gapi
   end
 
-  def query_job_json query, job_id: "job_9876543210", location: "US", dry_run: nil
+  def query_job_json query, job_id: "job_9876543210", location: "US", dry_run: nil, create_session: nil, session_id: nil
     hash = {
       "jobReference" => {
         "projectId" => project,
@@ -833,6 +836,7 @@ def query_job_json query, job_id: "job_9876543210", location: "US", dry_run: nil
           "defaultDataset" => nil,
           "destinationTable" => nil,
           "createDisposition" => nil,
+          "createSession" => create_session,
           "writeDisposition" => nil,
           "priority" => "INTERACTIVE",
           "allowLargeResults" => nil,
@@ -847,6 +851,7 @@ def query_job_json query, job_id: "job_9876543210", location: "US", dry_run: nil
       }
     }
     hash["jobReference"]["location"] = location if location
+    hash["configuration"]["query"]["connectionProperties"] = [{key: "session_id", value: session_id}] if session_id
     hash.to_json
   end