diff --git a/google-cloud-firestore/acceptance/firestore/collection_group_test.rb b/google-cloud-firestore/acceptance/firestore/collection_group_test.rb
new file mode 100644
index 000000000000..a4c30b353d25
--- /dev/null
+++ b/google-cloud-firestore/acceptance/firestore/collection_group_test.rb
@@ -0,0 +1,183 @@
+# Copyright 2021 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
+#
+#     https://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.
+
+require "firestore_helper"
+
+describe Google::Cloud::Firestore::CollectionGroup, :firestore_acceptance do
+  describe "#get" do
+    it "queries a collection group" do
+      collection_id = "b-#{SecureRandom.hex(4)}"
+      doc_paths = [
+          "abc/123/#{collection_id}/cg-doc1",
+          "abc/123/#{collection_id}/cg-doc2",
+          "#{collection_id}/cg-doc3",
+          "#{collection_id}/cg-doc4",
+          "def/456/#{collection_id}/cg-doc5",
+          "#{collection_id}/virtual-doc/nested-coll/not-cg-doc",
+          "x#{collection_id}/not-cg-doc",
+          "#{collection_id}x/not-cg-doc",
+          "abc/123/#{collection_id}x/not-cg-doc",
+          "abc/123/x#{collection_id}/not-cg-doc",
+          "abc/#{collection_id}"
+      ]
+      firestore.batch do |b|
+        doc_paths.each do |doc_path|
+          doc_ref = firestore.document doc_path
+          b.set doc_ref, {x: 1}
+        end
+      end
+
+      collection_group = firestore.collection_group(collection_id)
+      snapshots = collection_group.get
+      _(snapshots.map(&:document_id)).must_equal ["cg-doc1", "cg-doc2", "cg-doc3", "cg-doc4", "cg-doc5"]
+    end
+
+    it "queries a collection group with start_at and end_at" do
+      collection_id = "b-#{SecureRandom.hex(4)}"
+      doc_paths = [
+        "a/a/#{collection_id}/cg-doc1",
+        "a/b/a/b/#{collection_id}/cg-doc2",
+        "a/b/#{collection_id}/cg-doc3",
+        "a/b/c/d/#{collection_id}/cg-doc4",
+        "a/c/#{collection_id}/cg-doc5",
+        "#{collection_id}/cg-doc6",
+        "a/b/nope/nope"
+      ]
+      firestore.batch do |b|
+        doc_paths.each do |doc_path|
+          doc_ref = firestore.document doc_path
+          b.set doc_ref, {x: 1}
+        end
+      end
+
+      collection_group = firestore.collection_group(collection_id)
+        .order_by("__name__")
+        .start_at(firestore.document("a/b"))
+        .end_at(firestore.document("a/b0"))
+
+      snapshots = collection_group.get
+      _(snapshots.map(&:document_id)).must_equal ["cg-doc2", "cg-doc3", "cg-doc4"]
+
+      collection_group = firestore.collection_group(collection_id)
+        .order_by("__name__")
+        .start_after(firestore.document("a/b"))
+        .end_before(firestore.document("a/b/#{collection_id}/cg-doc3"))
+      snapshots = collection_group.get
+      _(snapshots.map(&:document_id)).must_equal ["cg-doc2"]
+    end
+
+    it "queries a collection group with filters" do
+      collection_id = "b-#{SecureRandom.hex(4)}"
+      doc_paths = [
+        "a/a/#{collection_id}/cg-doc1",
+        "a/b/a/b/#{collection_id}/cg-doc2",
+        "a/b/#{collection_id}/cg-doc3",
+        "a/b/c/d/#{collection_id}/cg-doc4",
+        "a/c/#{collection_id}/cg-doc5",
+        "#{collection_id}/cg-doc6",
+        "a/b/nope/nope"
+      ]
+      firestore.batch do |b|
+        doc_paths.each do |doc_path|
+          doc_ref = firestore.document doc_path
+          b.set doc_ref, {x: 1}
+        end
+      end
+
+      collection_group = firestore.collection_group(collection_id)
+        .where("__name__", ">=", firestore.document("a/b"))
+        .where("__name__", "<=", firestore.document("a/b0"))
+
+      snapshots = collection_group.get
+      _(snapshots.map(&:document_id)).must_equal ["cg-doc2", "cg-doc3", "cg-doc4"]
+
+      collection_group = firestore.collection_group(collection_id)
+        .where("__name__", ">", firestore.document("a/b"))
+        .where(
+          "__name__", "<", firestore.document("a/b/#{collection_id}/cg-doc3")
+        )
+      snapshots = collection_group.get
+      _(snapshots.map(&:document_id)).must_equal ["cg-doc2"]
+    end
+  end
+
+  describe "#partitions" do
+    it "queries a collection group using partitions" do
+      rand_col = firestore.col "#{root_path}/query/#{SecureRandom.hex(4)}"
+
+      document_ids = ["a", "b", "c"].map do |prefix|
+        # Minimum partition size is 128.
+        128.times.map do |i|
+          "#{prefix}#{(i+1).to_s.rjust(3, '0')}"
+        end
+      end.flatten # "a001", "a002", ... "c128"
+      firestore.batch do |b|
+        document_ids.each do |id|
+          doc_ref = rand_col.document id
+          b.set doc_ref, { foo: id }
+        end
+      end
+
+      collection_group = firestore.collection_group(rand_col.collection_id)
+
+      partitions = collection_group.partitions 6
+      _(partitions).must_be_kind_of Array
+      _(partitions.count).must_equal 3
+
+      _(partitions[0]).must_be_kind_of Google::Cloud::Firestore::QueryPartition
+      _(partitions[0].start_at).must_be :nil?
+      _(partitions[0].end_before).must_be_kind_of Array
+      _(partitions[0].end_before[0]).must_be_kind_of Google::Cloud::Firestore::DocumentReference
+      _(document_ids).must_include partitions[0].end_before[0].document_id
+
+      _(partitions[1]).must_be_kind_of Google::Cloud::Firestore::QueryPartition
+      _(partitions[1].start_at).must_be_kind_of Array
+      _(partitions[1].start_at[0]).must_be_kind_of Google::Cloud::Firestore::DocumentReference
+      _(document_ids).must_include partitions[1].start_at[0].document_id
+      _(partitions[1].end_before).must_be_kind_of Array
+      _(partitions[1].end_before[0]).must_be_kind_of Google::Cloud::Firestore::DocumentReference
+      _(document_ids).must_include partitions[1].end_before[0].document_id
+
+      # Verify that partitions are sorted ascending order
+      _(partitions[0].end_before[0].document_id).must_be :<, partitions[1].end_before[0].document_id
+
+      _(partitions[2]).must_be_kind_of Google::Cloud::Firestore::QueryPartition
+      _(partitions[2].start_at).must_be_kind_of Array
+      _(partitions[2].start_at[0]).must_be_kind_of Google::Cloud::Firestore::DocumentReference
+      _(document_ids).must_include partitions[2].start_at[0].document_id
+      _(partitions[2].end_before).must_be :nil?
+
+      queries = partitions.map(&:to_query)
+      _(queries.count).must_equal 3
+      results = queries.map do |query|
+        _(query).must_be_kind_of Google::Cloud::Firestore::Query
+        query.get.map do |snp|
+          _(snp).must_be_kind_of Google::Cloud::Firestore::DocumentSnapshot
+          snp.document_id
+        end
+      end
+      results.each { |result| _(result).wont_be :empty? }
+      # Verify all document IDs have been returned, in original order.
+      _(results.flatten).must_equal document_ids
+
+      # Verify QueryPartition#start_at and #end_before can be used with a new Query.
+      query = collection_group.order("__name__").start_at(partitions[1].start_at).end_before(partitions[1].end_before)
+      result = query.get.map do |snp|
+        _(snp).must_be_kind_of Google::Cloud::Firestore::DocumentSnapshot
+        snp.document_id
+      end
+      _(result).must_equal results[1]
+    end
+  end
+end
diff --git a/google-cloud-firestore/acceptance/firestore/query_test.rb b/google-cloud-firestore/acceptance/firestore/query_test.rb
index 9e318a7e1dd8..db837b36a478 100644
--- a/google-cloud-firestore/acceptance/firestore/query_test.rb
+++ b/google-cloud-firestore/acceptance/firestore/query_test.rb
@@ -285,100 +285,36 @@
     _(results.map { |doc| doc[:foo] }).must_equal ["a", "b"]
   end
 
-  describe "Collection Group" do
-    it "queries a collection group" do
-      collection_group = "b-#{SecureRandom.hex(4)}"
-      doc_paths = [
-          "abc/123/#{collection_group}/cg-doc1",
-          "abc/123/#{collection_group}/cg-doc2",
-          "#{collection_group}/cg-doc3",
-          "#{collection_group}/cg-doc4",
-          "def/456/#{collection_group}/cg-doc5",
-          "#{collection_group}/virtual-doc/nested-coll/not-cg-doc",
-          "x#{collection_group}/not-cg-doc",
-          "#{collection_group}x/not-cg-doc",
-          "abc/123/#{collection_group}x/not-cg-doc",
-          "abc/123/x#{collection_group}/not-cg-doc",
-          "abc/#{collection_group}"
-      ]
-      firestore.batch do |b|
-        doc_paths.each do |doc_path|
-          doc_ref = firestore.document doc_path
-          b.set doc_ref, {x: 1}
-        end
-      end
-
-      query = firestore.collection_group collection_group
-      snapshots = query.get
-      _(snapshots.map(&:document_id)).must_equal ["cg-doc1", "cg-doc2", "cg-doc3", "cg-doc4", "cg-doc5"]
-    end
-
-    it "queries a collection group with start_at and end_at" do
-      collection_group = "b-#{SecureRandom.hex(4)}"
-      doc_paths = [
-        "a/a/#{collection_group}/cg-doc1",
-        "a/b/a/b/#{collection_group}/cg-doc2",
-        "a/b/#{collection_group}/cg-doc3",
-        "a/b/c/d/#{collection_group}/cg-doc4",
-        "a/c/#{collection_group}/cg-doc5",
-        "#{collection_group}/cg-doc6",
-        "a/b/nope/nope"
-      ]
-      firestore.batch do |b|
-        doc_paths.each do |doc_path|
-          doc_ref = firestore.document doc_path
-          b.set doc_ref, {x: 1}
-        end
-      end
-
-      query = firestore.collection_group(collection_group)
-        .order_by("__name__")
-        .start_at(firestore.document("a/b"))
-        .end_at(firestore.document("a/b0"))
-
-      snapshots = query.get
-      _(snapshots.map(&:document_id)).must_equal ["cg-doc2", "cg-doc3", "cg-doc4"]
-
-      query = firestore.collection_group(collection_group)
-        .order_by("__name__")
-        .start_after(firestore.document("a/b"))
-        .end_before(firestore.document("a/b/#{collection_group}/cg-doc3"))
-      snapshots = query.get
-      _(snapshots.map(&:document_id)).must_equal ["cg-doc2"]
-    end
-
-    it "queries a collection group with filters" do
-      collection_group = "b-#{SecureRandom.hex(4)}"
-      doc_paths = [
-        "a/a/#{collection_group}/cg-doc1",
-        "a/b/a/b/#{collection_group}/cg-doc2",
-        "a/b/#{collection_group}/cg-doc3",
-        "a/b/c/d/#{collection_group}/cg-doc4",
-        "a/c/#{collection_group}/cg-doc5",
-        "#{collection_group}/cg-doc6",
-        "a/b/nope/nope"
-      ]
-      firestore.batch do |b|
-        doc_paths.each do |doc_path|
-          doc_ref = firestore.document doc_path
-          b.set doc_ref, {x: 1}
-        end
-      end
-
-      query = firestore.collection_group(collection_group)
-        .where("__name__", ">=", firestore.document("a/b"))
-        .where("__name__", "<=", firestore.document("a/b0"))
-
-      snapshots = query.get
-      _(snapshots.map(&:document_id)).must_equal ["cg-doc2", "cg-doc3", "cg-doc4"]
-
-      query = firestore.collection_group(collection_group)
-        .where("__name__", ">", firestore.document("a/b"))
-        .where(
-          "__name__", "<", firestore.document("a/b/#{collection_group}/cg-doc3")
-        )
-      snapshots = query.get
-      _(snapshots.map(&:document_id)).must_equal ["cg-doc2"]
-    end
+  it "has to_json method and from_json class method" do
+    rand_query_col = firestore.col "#{root_path}/query/#{SecureRandom.hex(4)}"
+    rand_query_col.doc("doc1").create({foo: "a"})
+    rand_query_col.doc("doc2").create({foo: "b"})
+    rand_query_col.doc("doc3").create({foo: "c"})
+
+    original_query = rand_query_col.order(:foo).limit_to_last 2
+
+    json = original_query.to_json
+    _(json).must_be_instance_of String
+
+    query = Google::Cloud::Firestore::Query.from_json json, firestore
+    _(query).must_be_instance_of Google::Cloud::Firestore::Query
+
+    results_1 = []
+    query.get { |result| results_1 << result } # block directly to get, rpc
+    _(results_1.map(&:document_id)).must_equal ["doc2","doc3"]
+    _(results_1.map { |doc| doc[:foo] }).must_equal ["b","c"]
+
+    results_2 = []
+    query.get { |result| results_2 << result } # block directly to get, rpc
+    _(results_2.map(&:document_id)).must_equal ["doc2","doc3"]
+    _(results_2.map { |doc| doc[:foo] }).must_equal ["b","c"]
+
+    results_3 = query.get # enum_for :get
+    _(results_3.map(&:document_id)).must_equal ["doc2","doc3"] # rpc
+    _(results_3.map { |doc| doc[:foo] }).must_equal ["b","c"] # rpc
+
+    results_4 = query.get # enum_for :get
+    _(results_4.map(&:document_id)).must_equal ["doc2","doc3"] # rpc
+    _(results_4.map { |doc| doc[:foo] }).must_equal ["b","c"] # rpc
   end
 end
diff --git a/google-cloud-firestore/acceptance/firestore/watch_test.rb b/google-cloud-firestore/acceptance/firestore/watch_test.rb
index c8b77cc6b49e..b385f641847c 100644
--- a/google-cloud-firestore/acceptance/firestore/watch_test.rb
+++ b/google-cloud-firestore/acceptance/firestore/watch_test.rb
@@ -188,7 +188,7 @@
   def wait_until &block
     wait_count = 0
     until block.call
-      fail "wait_until criterial was not met" if wait_count > 200
+      fail "wait_until criteria was not met" if wait_count > 200
       wait_count += 1
       sleep 0.01
     end
diff --git a/google-cloud-firestore/lib/google/cloud/firestore/client.rb b/google-cloud-firestore/lib/google/cloud/firestore/client.rb
index 737cfa4cd157..b6e8459074d3 100644
--- a/google-cloud-firestore/lib/google/cloud/firestore/client.rb
+++ b/google-cloud-firestore/lib/google/cloud/firestore/client.rb
@@ -20,6 +20,7 @@
 require "google/cloud/firestore/collection_reference"
 require "google/cloud/firestore/document_reference"
 require "google/cloud/firestore/document_snapshot"
+require "google/cloud/firestore/collection_group"
 require "google/cloud/firestore/batch"
 require "google/cloud/firestore/transaction"
 
@@ -139,7 +140,7 @@ def col collection_path
         alias collection col
 
         ##
-        # Creates and returns a new Query that includes all documents in the
+        # Creates and returns a new collection group that includes all documents in the
         # database that are contained in a collection or subcollection with the
         # given collection_id.
         #
@@ -147,7 +148,7 @@ def col collection_path
         #   over. Every collection or subcollection with this ID as the last
         #   segment of its path will be included. Cannot contain a slash (`/`).
         #
-        # @return [Query] The created Query.
+        # @return [CollectionGroup] The created collection group.
         #
         # @example
         #   require "google/cloud/firestore"
@@ -155,9 +156,9 @@ def col collection_path
         #   firestore = Google::Cloud::Firestore.new
         #
         #   # Get the cities collection group query
-        #   query = firestore.col_group "cities"
+        #   col_group = firestore.col_group "cities"
         #
-        #   query.get do |city|
+        #   col_group.get do |city|
         #     puts "#{city.document_id} has #{city[:population]} residents."
         #   end
         #
@@ -166,15 +167,8 @@ def col_group collection_id
             raise ArgumentError, "Invalid collection_id: '#{collection_id}', " \
               "must not contain '/'."
           end
-          query = Google::Cloud::Firestore::V1::StructuredQuery.new(
-            from: [
-              Google::Cloud::Firestore::V1::StructuredQuery::CollectionSelector.new(
-                collection_id: collection_id, all_descendants: true
-              )
-            ]
-          )
-
-          Query.start query, service.documents_path, self
+
+          CollectionGroup.from_collection_id service.documents_path, collection_id, self
         end
         alias collection_group col_group
 
diff --git a/google-cloud-firestore/lib/google/cloud/firestore/collection_group.rb b/google-cloud-firestore/lib/google/cloud/firestore/collection_group.rb
new file mode 100644
index 000000000000..ecc779188b89
--- /dev/null
+++ b/google-cloud-firestore/lib/google/cloud/firestore/collection_group.rb
@@ -0,0 +1,136 @@
+# Copyright 2021 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
+#
+#     https://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.
+
+
+require "google/cloud/firestore/v1"
+require "google/cloud/firestore/query"
+require "google/cloud/firestore/query_partition"
+
+module Google
+  module Cloud
+    module Firestore
+      ##
+      # # CollectionGroup
+      #
+      # A collection group object is used for adding documents, getting
+      # document references, and querying for documents, including with partitions.
+      #
+      # See {Client#col_group} and {Query}.
+      #
+      # @example
+      #   require "google/cloud/firestore"
+      #
+      #   firestore = Google::Cloud::Firestore.new
+      #
+      #   # Get a collection group
+      #   col_group = firestore.col_group "cities"
+      #
+      #   # Get and print all city documents
+      #   col_group.get do |city|
+      #     puts "#{city.document_id} has #{city[:population]} residents."
+      #   end
+      #
+      class CollectionGroup < Query
+        ##
+        # Partitions a query by returning partition cursors that can be used to run the query in parallel. The returned
+        # partition cursors are split points that can be used as starting/end points for the query results.
+        #
+        # @param [Integer] partition_count The desired maximum number of partition points. The number must be strictly
+        #   positive. The actual number of partitions returned may be fewer.
+        #
+        # @return [Array<QueryPartition>] An ordered array of query partitions.
+        #
+        # @example
+        #   require "google/cloud/firestore"
+        #
+        #   firestore = Google::Cloud::Firestore.new
+        #
+        #   col_group = firestore.col_group "cities"
+        #
+        #   partitions = col_group.partitions 3
+        #
+        #   queries = partitions.map(&:to_query)
+        #
+        def partitions partition_count
+          ensure_service!
+
+          raise ArgumentError, "partition_count must be > 0" unless partition_count.positive?
+
+          # Partition queries require explicit ordering by __name__.
+          query_with_default_order = order "__name__"
+          # Since we are always returning an extra partition (with en empty endBefore cursor), we reduce the desired
+          # partition count by one.
+          partition_count -= 1
+
+          grpc_partitions = if partition_count.positive?
+                              # Retrieve all pages, since cursor order is not guaranteed and they must be sorted.
+                              list_all partition_count, query_with_default_order
+                            else
+                              [] # Ensure that a single, empty QueryPartition is returned.
+                            end
+          cursor_values = grpc_partitions.map do |cursor|
+            # Convert each cursor to a (single-element) array of Google::Cloud::Firestore::DocumentReference.
+            cursor.values.map do |value|
+              Convert.value_to_raw value, client
+            end
+          end
+          # Sort the values of the returned cursor, which right now should only contain a single reference value (which
+          # needs to be sorted one component at a time).
+          cursor_values.sort! do |a, b|
+            a.first <=> b.first
+          end
+
+          start_at = nil
+          results = cursor_values.map do |end_before|
+            partition = QueryPartition.new query_with_default_order, start_at, end_before
+            start_at = end_before
+            partition
+          end
+          # Always add a final QueryPartition with an empty end_before value.
+          results << QueryPartition.new(query_with_default_order, start_at, nil)
+          results
+        end
+
+        ##
+        # @private New Collection group object from a path.
+        def self.from_collection_id parent_path, collection_id, client
+          query = Google::Cloud::Firestore::V1::StructuredQuery.new(
+            from: [
+              Google::Cloud::Firestore::V1::StructuredQuery::CollectionSelector.new(
+                collection_id: collection_id,
+                all_descendants: true
+              )
+            ]
+          )
+          CollectionGroup.new query, parent_path, client
+        end
+
+        protected
+
+        def list_all partition_count, query_with_default_order
+          grpc_partitions = []
+          token = nil
+          loop do
+            grpc = service.partition_query parent_path, query_with_default_order.query, partition_count, token: token
+            grpc_partitions += Array(grpc.partitions)
+            token = grpc.next_page_token
+            token = nil if token == ""
+            break unless token
+          end
+          grpc_partitions
+        end
+      end
+    end
+  end
+end
diff --git a/google-cloud-firestore/lib/google/cloud/firestore/collection_reference.rb b/google-cloud-firestore/lib/google/cloud/firestore/collection_reference.rb
index 3a8b7d3c97cf..684c2206548e 100644
--- a/google-cloud-firestore/lib/google/cloud/firestore/collection_reference.rb
+++ b/google-cloud-firestore/lib/google/cloud/firestore/collection_reference.rb
@@ -14,9 +14,9 @@
 
 
 require "google/cloud/firestore/v1"
+require "google/cloud/firestore/query"
 require "google/cloud/firestore/document_reference"
 require "google/cloud/firestore/document_snapshot"
-require "google/cloud/firestore/query"
 require "google/cloud/firestore/generate"
 require "google/cloud/firestore/collection_reference_list"
 
@@ -47,6 +47,13 @@ class CollectionReference < Query
         # @private The firestore client object.
         attr_accessor :client
 
+        ##
+        # @private Creates a new CollectionReference.
+        def initialize query, path, client
+          super query, nil, client # Pass nil parent_path arg since this class implements #parent_path
+          @path = path
+        end
+
         ##
         # The collection identifier for the collection resource.
         #
@@ -257,11 +264,7 @@ def self.from_path path, client
             ]
           )
 
-          new.tap do |c|
-            c.client = client
-            c.instance_variable_set :@path, path
-            c.instance_variable_set :@query, query
-          end
+          CollectionReference.new query, path, client
         end
 
         protected
diff --git a/google-cloud-firestore/lib/google/cloud/firestore/document_reference.rb b/google-cloud-firestore/lib/google/cloud/firestore/document_reference.rb
index c5da03ef4283..940230a88390 100644
--- a/google-cloud-firestore/lib/google/cloud/firestore/document_reference.rb
+++ b/google-cloud-firestore/lib/google/cloud/firestore/document_reference.rb
@@ -18,6 +18,7 @@
 require "google/cloud/firestore/collection_reference"
 require "google/cloud/firestore/document_listener"
 require "google/cloud/firestore/document_reference/list"
+require "google/cloud/firestore/resource_path"
 
 module Google
   module Cloud
@@ -459,6 +460,12 @@ def delete exists: nil, update_time: nil
 
         # @!endgroup
 
+        # @private
+        def <=> other
+          return nil unless other.is_a? self.class
+          ResourcePath.from_path(path) <=> ResourcePath.from_path(other.path)
+        end
+
         ##
         # @private New DocumentReference object from a path.
         def self.from_path path, client
diff --git a/google-cloud-firestore/lib/google/cloud/firestore/document_reference/list.rb b/google-cloud-firestore/lib/google/cloud/firestore/document_reference/list.rb
index 87ff400ca365..776cf76c8cf7 100644
--- a/google-cloud-firestore/lib/google/cloud/firestore/document_reference/list.rb
+++ b/google-cloud-firestore/lib/google/cloud/firestore/document_reference/list.rb
@@ -109,7 +109,7 @@ def next
           #
           # @return [Enumerator]
           #
-          # @example Iterating each document reference by passing a block:
+          # @example Iterating each document reference by passing a block or proc:
           #   require "google/cloud/firestore"
           #
           #   firestore = Google::Cloud::Firestore.new
diff --git a/google-cloud-firestore/lib/google/cloud/firestore/field_value.rb b/google-cloud-firestore/lib/google/cloud/firestore/field_value.rb
index 9a3855f51d93..97cd7ce661b2 100644
--- a/google-cloud-firestore/lib/google/cloud/firestore/field_value.rb
+++ b/google-cloud-firestore/lib/google/cloud/firestore/field_value.rb
@@ -27,9 +27,14 @@ module Firestore
       #
       #   firestore = Google::Cloud::Firestore.new
       #
-      #   user_snap = firestore.doc("users/frank").get
+      #   # Get a document reference
+      #   nyc_ref = firestore.doc "cities/NYC"
       #
-      #   # TODO
+      #   # Set the population to increment by 1.
+      #   increment_value = Google::Cloud::Firestore::FieldValue.increment 1
+      #
+      #   nyc_ref.update({ name: "New York City",
+      #                    population: increment_value })
       #
       class FieldValue
         ##
diff --git a/google-cloud-firestore/lib/google/cloud/firestore/query.rb b/google-cloud-firestore/lib/google/cloud/firestore/query.rb
index 757f83363dd9..13af07448109 100644
--- a/google-cloud-firestore/lib/google/cloud/firestore/query.rb
+++ b/google-cloud-firestore/lib/google/cloud/firestore/query.rb
@@ -17,6 +17,7 @@
 require "google/cloud/firestore/document_snapshot"
 require "google/cloud/firestore/query_listener"
 require "google/cloud/firestore/convert"
+require "json"
 
 module Google
   module Cloud
@@ -74,6 +75,16 @@ class Query
         # @private The firestore client object.
         attr_accessor :client
 
+        ##
+        # @private Creates a new Query.
+        def initialize query, parent_path, client, limit_type: nil
+          query ||= StructuredQuery.new
+          @query = query
+          @parent_path = parent_path
+          @limit_type = limit_type
+          @client = client
+        end
+
         ##
         # Restricts documents matching the query to return only data for the
         # provided fields.
@@ -962,16 +973,71 @@ def listen &callback
         end
         alias on_snapshot listen
 
+        ##
+        # Serializes the instance to a JSON text string. See also {Query.from_json}.
+        #
+        # @return [String] A JSON text string.
+        #
+        # @example
+        #   require "google/cloud/firestore"
+        #
+        #   firestore = Google::Cloud::Firestore.new
+        #   query = firestore.col(:cities).select(:population)
+        #
+        #   json = query.to_json
+        #
+        #   new_query = Google::Cloud::Firestore::Query.from_json json, firestore
+        #
+        #   new_query.get do |city|
+        #     puts "#{city.document_id} has #{city[:population]} residents."
+        #   end
+        #
+        def to_json options = nil
+          query_json = Google::Cloud::Firestore::V1::StructuredQuery.encode_json query
+          {
+            "query" => JSON.parse(query_json),
+            "parent_path" => parent_path,
+            "limit_type" => limit_type
+          }.to_json options
+        end
+
+        ##
+        # Deserializes a JSON text string serialized from this class and returns it as a new instance. See also
+        # {#to_json}.
+        #
+        # @param [String] json A JSON text string serialized using {#to_json}.
+        # @param [Google::Cloud::Firestore::Client] client A connected client instance.
+        #
+        # @return [Query] A new query equal to the original query used to create the JSON text string.
+        #
+        # @example
+        #   require "google/cloud/firestore"
+        #
+        #   firestore = Google::Cloud::Firestore.new
+        #   query = firestore.col(:cities).select(:population)
+        #
+        #   json = query.to_json
+        #
+        #   new_query = Google::Cloud::Firestore::Query.from_json json, firestore
+        #
+        #   new_query.get do |city|
+        #     puts "#{city.document_id} has #{city[:population]} residents."
+        #   end
+        #
+        def self.from_json json, client
+          raise ArgumentError, "client is required" unless client
+
+          json = JSON.parse json
+          query_json = json["query"]
+          raise ArgumentError, "Field 'query' is required" unless query_json
+          query = Google::Cloud::Firestore::V1::StructuredQuery.decode_json query_json.to_json
+          start query, json["parent_path"], client, limit_type: json["limit_type"]&.to_sym
+        end
+
         ##
         # @private Start a new Query.
         def self.start query, parent_path, client, limit_type: nil
-          query ||= StructuredQuery.new
-          Query.new.tap do |q|
-            q.instance_variable_set :@query, query
-            q.instance_variable_set :@parent_path, parent_path
-            q.instance_variable_set :@limit_type, limit_type
-            q.instance_variable_set :@client, client
-          end
+          new query, parent_path, client, limit_type: limit_type
         end
 
         protected
@@ -1094,11 +1160,14 @@ def values_to_cursor values, query
             return snapshot_to_cursor values.first, query
           end
 
+          # The *values param in start_at, start_after, etc. will wrap an array argument in an array, so unwrap it here.
+          values = values.first if values.count == 1 && values.first.is_a?(Array)
+
           # pair values with their field_paths to ensure correct formatting
           order_field_paths = order_by_field_paths query
           if values.count > order_field_paths.count
             # raise if too many values provided for the cursor
-            raise ArgumentError, "too many values"
+            raise ArgumentError, "There cannot be more cursor values than order by fields"
           end
 
           values = values.zip(order_field_paths).map do |value, field_path|
@@ -1130,7 +1199,6 @@ def snapshot_to_cursor snapshot, query
               snapshot[field_path]
             end
           end
-
           values_to_cursor values, query
         end
 
diff --git a/google-cloud-firestore/lib/google/cloud/firestore/query_partition.rb b/google-cloud-firestore/lib/google/cloud/firestore/query_partition.rb
new file mode 100644
index 000000000000..362e54ec8590
--- /dev/null
+++ b/google-cloud-firestore/lib/google/cloud/firestore/query_partition.rb
@@ -0,0 +1,80 @@
+# Copyright 2021 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
+#
+#     https://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.
+
+
+module Google
+  module Cloud
+    module Firestore
+      ##
+      # # QueryPartition
+      #
+      # Represents a split point that can be used in a query as a starting and/or end point for the query results.
+      #
+      # The cursors returned by {#start_at} and {#end_before} can only be used in a query that matches the constraint of
+      # the query that produced this partition.
+      #
+      # See {CollectionGroup#partitions} and {Query}.
+      #
+      # @!attribute [r] start_at
+      #   The cursor values that define the first result for this partition, or `nil` if this is the first partition.
+      #   Returns an array of values that represent a position, in the order they appear in the order by clause of the
+      #   query. Can contain fewer values than specified in the order by clause. Will be used in the query returned by
+      #   {#to_query}.
+      #   @return [Array<Object>, nil] Typically, the values are {DocumentReference} objects.
+      # @!attribute [r] end_before
+      #   The cursor values that define the first result after this partition, or `nil` if this is the last partition.
+      #   Returns an array of values that represent a position, in the order they appear in the order by clause of the
+      #   query.  Can contain fewer values than specified in the order by clause. Will be used in the query returned by
+      #   {#to_query}.
+      #   @return [Array<Object>, nil] Typically, the values are {DocumentReference} objects.
+      #
+      # @example
+      #   require "google/cloud/firestore"
+      #
+      #   firestore = Google::Cloud::Firestore.new
+      #
+      #   col_group = firestore.col_group "cities"
+      #
+      #   partitions = col_group.partitions 3
+      #
+      #   queries = partitions.map(&:to_query)
+      #
+      class QueryPartition
+        attr_reader :start_at
+        attr_reader :end_before
+
+        ##
+        # @private New QueryPartition from query and Cursor
+        def initialize query, start_at, end_before
+          @query = query
+          @start_at = start_at
+          @end_before = end_before
+        end
+
+        ##
+        # Creates a new query that only returns the documents for this partition, using the cursor values from
+        # {#start_at} and {#end_before}.
+        #
+        # @return [Query] The query for the partition.
+        #
+        def to_query
+          base_query = @query
+          base_query = base_query.start_at start_at if start_at
+          base_query = base_query.end_before end_before if end_before
+          base_query
+        end
+      end
+    end
+  end
+end
diff --git a/google-cloud-firestore/lib/google/cloud/firestore/resource_path.rb b/google-cloud-firestore/lib/google/cloud/firestore/resource_path.rb
new file mode 100644
index 000000000000..c5d4d1fb0e14
--- /dev/null
+++ b/google-cloud-firestore/lib/google/cloud/firestore/resource_path.rb
@@ -0,0 +1,58 @@
+# Copyright 2021 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
+#
+#     https://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.
+
+
+module Google
+  module Cloud
+    module Firestore
+      ##
+      # @private
+      #
+      # Represents a resource path to the Firestore API.
+      #
+      class ResourcePath
+        include Comparable
+
+        RESOURCE_PATH_RE = %r{^projects/([^/]*)/databases/([^/]*)(?:/documents/)?([\s\S]*)$}.freeze
+
+        attr_reader :project_id
+        attr_reader :database_id
+        attr_reader :segments
+
+        ##
+        # Creates a resource path object.
+        #
+        # @param [Array<String>] segments One or more strings representing the resource path.
+        #
+        # @return [ResourcePath] The resource path object.
+        #
+        def initialize project_id, database_id, segments
+          @project_id = project_id
+          @database_id = database_id
+          @segments = segments.split "/"
+        end
+
+        def <=> other
+          return nil unless other.is_a? ResourcePath
+          [project_id, database_id, segments] <=> [other.project_id, other.database_id, other.segments]
+        end
+
+        def self.from_path path
+          data = RESOURCE_PATH_RE.match path
+          new data[1], data[2], data[3]
+        end
+      end
+    end
+  end
+end
diff --git a/google-cloud-firestore/lib/google/cloud/firestore/service.rb b/google-cloud-firestore/lib/google/cloud/firestore/service.rb
index afc2fec7f628..54c4acf7d77b 100644
--- a/google-cloud-firestore/lib/google/cloud/firestore/service.rb
+++ b/google-cloud-firestore/lib/google/cloud/firestore/service.rb
@@ -96,6 +96,20 @@ def list_collections parent, token: nil, max: nil
           )
         end
 
+        ##
+        # Returns Google::Cloud::Firestore::V1::PartitionQueryResponse
+        def partition_query parent, query_grpc, partition_count, token: nil, max: nil
+          request = Google::Cloud::Firestore::V1::PartitionQueryRequest.new(
+            parent: parent,
+            structured_query: query_grpc,
+            partition_count: partition_count,
+            page_token: token,
+            page_size: max
+          )
+          paged_enum = firestore.partition_query request
+          paged_enum.response
+        end
+
         def run_query path, query_grpc, transaction: nil
           run_query_req = {
             parent:           path,
diff --git a/google-cloud-firestore/support/doctest_helper.rb b/google-cloud-firestore/support/doctest_helper.rb
index baef8699aa44..803fe40606b9 100644
--- a/google-cloud-firestore/support/doctest_helper.rb
+++ b/google-cloud-firestore/support/doctest_helper.rb
@@ -178,6 +178,18 @@ def mock_firestore
     end
   end
 
+  doctest.before "Google::Cloud::Firestore::CollectionGroup" do
+    mock_firestore do |mock|
+      mock.expect :run_query, run_query_resp, run_query_args
+    end
+  end
+
+  doctest.before "Google::Cloud::Firestore::CollectionGroup#partitions" do
+    mock_firestore do |mock|
+      mock.expect :partition_query, partition_query_resp, [Google::Cloud::Firestore::V1::PartitionQueryRequest]
+    end
+  end
+
   doctest.skip "Google::Cloud::Firestore::FieldPath" do
     mock_firestore do |mock|
       mock.expect :batch_get_documents, batch_get_resp_users, batch_get_args
@@ -286,6 +298,12 @@ def mock_firestore
     end
   end
 
+  doctest.before "Google::Cloud::Firestore::QueryPartition" do
+    mock_firestore do |mock|
+      mock.expect :partition_query, partition_query_resp, [Google::Cloud::Firestore::V1::PartitionQueryRequest]
+    end
+  end
+
   doctest.before "Google::Cloud::Firestore::CollectionReference" do
     mock_firestore do |mock|
       mock.expect :run_query, run_query_resp, run_query_args
@@ -499,6 +517,10 @@ def list_documents_args
   ]
 end
 
+def document_path doc_id
+  "projects/my-project-id/databases/(default)/documents/#{doc_id}"
+end
+
 def document_gapi doc: "my-document", fields: {}
   Google::Cloud::Firestore::V1::Document.new(
     name: "projects/my-project-id/databases/(default)/documents/#{doc}",
@@ -515,3 +537,25 @@ def documents_resp token: nil
   response.next_page_token = token if token
   paged_enum_struct response
 end
+
+def partition_query_resp count: 3, token: nil
+  response = Google::Cloud::Firestore::V1::PartitionQueryResponse.new(
+    # Minimum partition size is 128.
+    partitions: count.times.map { |i| cursor_grpc doc_ids: [((i+1) * 10).to_s] }
+  )
+  response.next_page_token = token if token
+  paged_enum_struct response
+end
+
+# Minimum partition size is 128.
+def cursor_grpc doc_ids: ["10"], before: true
+  converted_values = doc_ids.map do |doc_id|
+    Google::Cloud::Firestore::V1::Value.new(
+      reference_value: document_path(doc_id)
+    )
+  end
+  Google::Cloud::Firestore::V1::Cursor.new(
+    values: converted_values,
+    before: before
+  )
+end
diff --git a/google-cloud-firestore/test/google/cloud/firestore/client/col_group_test.rb b/google-cloud-firestore/test/google/cloud/firestore/client/col_group_test.rb
index 273252f45157..800961e43d47 100644
--- a/google-cloud-firestore/test/google/cloud/firestore/client/col_group_test.rb
+++ b/google-cloud-firestore/test/google/cloud/firestore/client/col_group_test.rb
@@ -19,10 +19,10 @@
   let(:collection_id_bad) { "a/b/my-collection-id" }
 
   it "creates a collection group query" do
-    query = firestore.col_group(collection_id).where "foo", "==", "bar"
+    collection_group = firestore.col_group(collection_id)
 
-    _(query).must_be_kind_of Google::Cloud::Firestore::Query
-    query_gapi = query.query
+    _(collection_group).must_be_kind_of Google::Cloud::Firestore::Query
+    query_gapi = collection_group.query
     _(query_gapi).must_be_kind_of Google::Cloud::Firestore::V1::StructuredQuery
     _(query_gapi.from.size).must_equal 1
     _(query_gapi.from.first).must_be_kind_of Google::Cloud::Firestore::V1::StructuredQuery::CollectionSelector
@@ -31,10 +31,10 @@
   end
 
   it "creates a collection group query using collection_group alias" do
-    query = firestore.collection_group(collection_id).where "foo", "==", "bar"
+    collection_group = firestore.collection_group(collection_id)
 
-    _(query).must_be_kind_of Google::Cloud::Firestore::Query
-    query_gapi = query.query
+    _(collection_group).must_be_kind_of Google::Cloud::Firestore::Query
+    query_gapi = collection_group.query
     _(query_gapi).must_be_kind_of Google::Cloud::Firestore::V1::StructuredQuery
     _(query_gapi.from.size).must_equal 1
     _(query_gapi.from.first).must_be_kind_of Google::Cloud::Firestore::V1::StructuredQuery::CollectionSelector
diff --git a/google-cloud-firestore/test/google/cloud/firestore/collection_group_test.rb b/google-cloud-firestore/test/google/cloud/firestore/collection_group_test.rb
new file mode 100644
index 000000000000..b39bda61278e
--- /dev/null
+++ b/google-cloud-firestore/test/google/cloud/firestore/collection_group_test.rb
@@ -0,0 +1,127 @@
+# Copyright 2021 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
+#
+#     https://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.
+
+require "helper"
+
+describe Google::Cloud::Firestore::CollectionGroup, :mock_firestore do
+  let(:collection_id) { "my-collection-id" }
+  let(:collection_group) do
+    Google::Cloud::Firestore::CollectionGroup.from_collection_id documents_path, collection_id, firestore
+  end
+  let(:expected_query) { collection_group_query }
+
+  it "raises if partition_count is < 1" do
+    expect do
+      partitions = collection_group.partitions 0
+    end.must_raise ArgumentError
+  end
+
+  it "returns 1 empty partition if partition_count is 1, without RPC" do
+    partitions = collection_group.partitions 1
+
+    _(partitions).must_be_kind_of Array
+    _(partitions.count).must_equal 1
+
+    _(partitions[0]).must_be_kind_of Google::Cloud::Firestore::QueryPartition
+    _(partitions[0].start_at).must_be :nil?
+    _(partitions[0].end_before).must_be :nil?
+  end
+
+  it "returns 1 empty partition if RPC returns no partitions" do
+    list_res = paged_enum_struct partition_query_resp(doc_ids: [])
+    firestore_mock.expect :partition_query, list_res, partition_query_args(expected_query)
+
+    partitions = collection_group.partitions 3
+
+    firestore_mock.verify
+
+    _(partitions).must_be_kind_of Array
+    _(partitions.count).must_equal 1
+
+    _(partitions[0]).must_be_kind_of Google::Cloud::Firestore::QueryPartition
+    _(partitions[0].start_at).must_be :nil?
+    _(partitions[0].end_before).must_be :nil?
+  end
+
+  it "sorts and lists partitions" do
+    # Results should be sorted so that "alice" comes before "alice-"
+    # Use an ID ending in "-" to ensure correct sorting, since full path strings are sorted dash before slash
+    # See Google::Cloud::Firestore::ResourcePath
+    list_res = paged_enum_struct partition_query_resp(doc_ids: ["alice-", "alice"])
+    firestore_mock.expect :partition_query, list_res, partition_query_args(expected_query)
+
+    partitions = collection_group.partitions 3
+
+    firestore_mock.verify
+
+    _(partitions).must_be_kind_of Array
+    _(partitions.count).must_equal 3
+
+    _(partitions[0]).must_be_kind_of Google::Cloud::Firestore::QueryPartition
+    _(partitions[0].start_at).must_be :nil?
+    _(partitions[0].end_before).must_be_kind_of Array
+    _(partitions[0].end_before.count).must_equal 1 # The array should have an element for each field in Order By.
+    _(partitions[0].end_before[0]).must_be_kind_of Google::Cloud::Firestore::DocumentReference
+    _(partitions[0].end_before[0].path).must_equal document_path("alice")
+
+    _(partitions[1]).must_be_kind_of Google::Cloud::Firestore::QueryPartition
+    _(partitions[1].start_at).must_be_kind_of Array
+    _(partitions[1].start_at.count).must_equal 1 # The array should have an element for each field in Order By.
+    _(partitions[1].start_at[0]).must_be_kind_of Google::Cloud::Firestore::DocumentReference
+    _(partitions[1].start_at[0].path).must_equal document_path("alice")
+    _(partitions[1].end_before).must_be_kind_of Array
+    _(partitions[1].end_before.count).must_equal 1 # The array should have an element for each field in Order By.
+    _(partitions[1].end_before[0]).must_be_kind_of Google::Cloud::Firestore::DocumentReference
+    _(partitions[1].end_before[0].path).must_equal document_path("alice-")
+
+    _(partitions[2]).must_be_kind_of Google::Cloud::Firestore::QueryPartition
+    _(partitions[2].start_at).must_be_kind_of Array
+    _(partitions[2].start_at.count).must_equal 1 # The array should have an element for each field in Order By.
+    _(partitions[2].start_at[0]).must_be_kind_of Google::Cloud::Firestore::DocumentReference
+    _(partitions[2].start_at[0].path).must_equal document_path("alice-")
+    _(partitions[2].end_before).must_be :nil?
+
+    query_1 = partitions[0].to_query
+    _(query_1).must_be_kind_of Google::Cloud::Firestore::Query
+    _(query_1.query).must_equal collection_group_query(end_before: ["alice"])
+
+    query_2 = partitions[1].to_query
+    _(query_2).must_be_kind_of Google::Cloud::Firestore::Query
+    _(query_2.query).must_equal collection_group_query(start_at: ["alice"], end_before: ["alice-"])
+
+    query_3 = partitions[2].to_query
+    _(query_3).must_be_kind_of Google::Cloud::Firestore::Query
+    _(query_3.query).must_equal collection_group_query(start_at: ["alice-"])
+  end
+
+  def collection_group_query start_at: nil, end_before: nil
+    query_grpc = Google::Cloud::Firestore::V1::StructuredQuery.new(
+      from: [
+        Google::Cloud::Firestore::V1::StructuredQuery::CollectionSelector.new(
+          collection_id: "my-collection-id",
+          all_descendants: true
+        )
+      ],
+      order_by: [
+        Google::Cloud::Firestore::V1::StructuredQuery::Order.new(
+          field: Google::Cloud::Firestore::V1::StructuredQuery::FieldReference.new(field_path: "__name__"),
+          direction: :ASCENDING
+        )
+      ]
+    )
+    query_grpc.start_at = cursor_grpc(doc_ids: start_at) if start_at
+    query_grpc.end_at = cursor_grpc(doc_ids: end_before) if end_before
+    query_grpc
+  end
+end
diff --git a/google-cloud-firestore/test/google/cloud/firestore/collection_reference/list_documents_test.rb b/google-cloud-firestore/test/google/cloud/firestore/collection_reference/list_documents_test.rb
index a73a097f7e78..9de697a5bf41 100644
--- a/google-cloud-firestore/test/google/cloud/firestore/collection_reference/list_documents_test.rb
+++ b/google-cloud-firestore/test/google/cloud/firestore/collection_reference/list_documents_test.rb
@@ -137,21 +137,6 @@
     _(documents.token).must_equal "next_page_token"
   end
 
-  it "paginates documents without max set" do
-    list_res = paged_enum_struct list_documents_gapi(3, "next_page_token")
-
-    firestore_mock.expect :list_documents, list_res, list_documents_args
-
-    documents = collection.list_documents
-
-    firestore_mock.verify
-
-    documents.each { |m| _(m).must_be_kind_of Google::Cloud::Firestore::DocumentReference }
-    _(documents.count).must_equal 3
-    _(documents.token).wont_be :nil?
-    _(documents.token).must_equal "next_page_token"
-  end
-
   def document_gapi
     Google::Cloud::Firestore::V1::Document.new(
       name: "projects/#{project}/databases/(default)/documents/my-document",
diff --git a/google-cloud-firestore/test/google/cloud/firestore/query/cursors_test.rb b/google-cloud-firestore/test/google/cloud/firestore/query/cursors_test.rb
index a9346b8c3e2c..61648762d57a 100644
--- a/google-cloud-firestore/test/google/cloud/firestore/query/cursors_test.rb
+++ b/google-cloud-firestore/test/google/cloud/firestore/query/cursors_test.rb
@@ -305,7 +305,40 @@
     _(generated_query).must_equal expected_query
   end
 
-  it "Start/End with two values" do
+  it "StartAt/EndBefore with two values" do
+    expected_query = Google::Cloud::Firestore::V1::StructuredQuery.new(
+      from: [Google::Cloud::Firestore::V1::StructuredQuery::CollectionSelector.new(collection_id: "C")],
+      order_by: [
+        Google::Cloud::Firestore::V1::StructuredQuery::Order.new(
+          field: Google::Cloud::Firestore::V1::StructuredQuery::FieldReference.new(field_path: "a"),
+          direction: :ASCENDING
+        ),
+        Google::Cloud::Firestore::V1::StructuredQuery::Order.new(
+          field: Google::Cloud::Firestore::V1::StructuredQuery::FieldReference.new(field_path: "b"),
+          direction: :DESCENDING
+        )
+      ],
+      start_at: Google::Cloud::Firestore::V1::Cursor.new(
+        values: [
+          Google::Cloud::Firestore::Convert.raw_to_value(7),
+          Google::Cloud::Firestore::Convert.raw_to_value(8)
+        ],
+        before: true
+      ),
+      end_at: Google::Cloud::Firestore::V1::Cursor.new(
+        values: [
+          Google::Cloud::Firestore::Convert.raw_to_value(9),
+          Google::Cloud::Firestore::Convert.raw_to_value(10)
+        ],
+        before: true
+      )
+    )
+
+    generated_query = collection.order(:a).order(:b, :desc).start_at(7, 8).end_before(9, 10).query
+    _(generated_query).must_equal expected_query
+  end
+
+  it "StartAfter/EndAt with two values" do
     expected_query = Google::Cloud::Firestore::V1::StructuredQuery.new(
       from: [Google::Cloud::Firestore::V1::StructuredQuery::CollectionSelector.new(collection_id: "C")],
       order_by: [
@@ -338,6 +371,72 @@
     _(generated_query).must_equal expected_query
   end
 
+  it "StartAt/EndBefore with array values" do
+    expected_query = Google::Cloud::Firestore::V1::StructuredQuery.new(
+      from: [Google::Cloud::Firestore::V1::StructuredQuery::CollectionSelector.new(collection_id: "C")],
+      order_by: [
+        Google::Cloud::Firestore::V1::StructuredQuery::Order.new(
+          field: Google::Cloud::Firestore::V1::StructuredQuery::FieldReference.new(field_path: "a"),
+          direction: :ASCENDING
+        ),
+        Google::Cloud::Firestore::V1::StructuredQuery::Order.new(
+          field: Google::Cloud::Firestore::V1::StructuredQuery::FieldReference.new(field_path: "b"),
+          direction: :DESCENDING
+        )
+      ],
+      start_at: Google::Cloud::Firestore::V1::Cursor.new(
+        values: [
+          Google::Cloud::Firestore::Convert.raw_to_value(7),
+          Google::Cloud::Firestore::Convert.raw_to_value(8)
+        ],
+        before: true
+      ),
+      end_at: Google::Cloud::Firestore::V1::Cursor.new(
+        values: [
+          Google::Cloud::Firestore::Convert.raw_to_value(9),
+          Google::Cloud::Firestore::Convert.raw_to_value(10)
+        ],
+        before: true
+      )
+    )
+
+    generated_query = collection.order(:a).order(:b, :desc).start_at([7, 8]).end_before([9, 10]).query
+    _(generated_query).must_equal expected_query
+  end
+
+  it "StartAfter/EndAt with array values" do
+    expected_query = Google::Cloud::Firestore::V1::StructuredQuery.new(
+      from: [Google::Cloud::Firestore::V1::StructuredQuery::CollectionSelector.new(collection_id: "C")],
+      order_by: [
+        Google::Cloud::Firestore::V1::StructuredQuery::Order.new(
+          field: Google::Cloud::Firestore::V1::StructuredQuery::FieldReference.new(field_path: "a"),
+          direction: :ASCENDING
+        ),
+        Google::Cloud::Firestore::V1::StructuredQuery::Order.new(
+          field: Google::Cloud::Firestore::V1::StructuredQuery::FieldReference.new(field_path: "b"),
+          direction: :DESCENDING
+        )
+      ],
+      start_at: Google::Cloud::Firestore::V1::Cursor.new(
+        values: [
+          Google::Cloud::Firestore::Convert.raw_to_value(7),
+          Google::Cloud::Firestore::Convert.raw_to_value(8)
+        ],
+        before: false
+      ),
+      end_at: Google::Cloud::Firestore::V1::Cursor.new(
+        values: [
+          Google::Cloud::Firestore::Convert.raw_to_value(9),
+          Google::Cloud::Firestore::Convert.raw_to_value(10)
+        ],
+        before: false
+      )
+    )
+
+    generated_query = collection.order(:a).order(:b, :desc).start_after([7, 8]).end_at([9, 10]).query
+    _(generated_query).must_equal expected_query
+  end
+
   it "with __name__" do
     # TODO: Looks like we need to create the full path when paired with __name__
     expected_query = Google::Cloud::Firestore::V1::StructuredQuery.new(
diff --git a/google-cloud-firestore/test/google/cloud/firestore/query/get_test.rb b/google-cloud-firestore/test/google/cloud/firestore/query/get_test.rb
index 29d2a6d6fb0a..4def253dbf5e 100644
--- a/google-cloud-firestore/test/google/cloud/firestore/query/get_test.rb
+++ b/google-cloud-firestore/test/google/cloud/firestore/query/get_test.rb
@@ -253,6 +253,43 @@
     assert_results_enum results_enum
   end
 
+  it "gets a complex query after serialization and deserialization" do
+    expected_query = Google::Cloud::Firestore::V1::StructuredQuery.new(
+      select: Google::Cloud::Firestore::V1::StructuredQuery::Projection.new(
+        fields: [Google::Cloud::Firestore::V1::StructuredQuery::FieldReference.new(field_path: "name")]),
+      from: [Google::Cloud::Firestore::V1::StructuredQuery::CollectionSelector.new(collection_id: "users", all_descendants: false)],
+      offset: 3,
+      limit: Google::Protobuf::Int32Value.new(value: 42),
+      order_by: [
+        Google::Cloud::Firestore::V1::StructuredQuery::Order.new(
+          field: Google::Cloud::Firestore::V1::StructuredQuery::FieldReference.new(field_path: "name"),
+          direction: :ASCENDING),
+        Google::Cloud::Firestore::V1::StructuredQuery::Order.new(
+          field: Google::Cloud::Firestore::V1::StructuredQuery::FieldReference.new(field_path: "__name__"),
+          direction: :DESCENDING)],
+      start_at: Google::Cloud::Firestore::V1::Cursor.new(values: [Google::Cloud::Firestore::Convert.raw_to_value("foo")], before: false),
+      end_at: Google::Cloud::Firestore::V1::Cursor.new(values: [Google::Cloud::Firestore::Convert.raw_to_value("bar")], before: true)
+    )
+    firestore_mock.expect :run_query, query_results_enum, run_query_args(expected_query)
+
+    original_query = firestore.col(:users).select(:name).offset(3).limit(42).order(:name).order(firestore.document_id, :desc).start_after(:foo).end_before(:bar)
+
+    json = original_query.to_json
+    _(json).must_be_instance_of String
+
+    deserialized_query = Google::Cloud::Firestore::Query.from_json json, firestore
+    _(deserialized_query).must_be_instance_of Google::Cloud::Firestore::Query
+
+    _(deserialized_query.query).must_equal expected_query # Private field
+    _(deserialized_query.parent_path).must_equal original_query.parent_path # Private field
+    _(deserialized_query.limit_type).must_equal original_query.limit_type # Private field
+    _(deserialized_query.client).must_equal original_query.client # Private field
+
+    results_enum = deserialized_query.get
+
+    assert_results_enum results_enum
+  end
+
   def assert_results_enum enum
     _(enum).must_be_kind_of Enumerator
 
diff --git a/google-cloud-firestore/test/google/cloud/firestore/resource_path_test.rb b/google-cloud-firestore/test/google/cloud/firestore/resource_path_test.rb
new file mode 100644
index 000000000000..34bdb70ee8fe
--- /dev/null
+++ b/google-cloud-firestore/test/google/cloud/firestore/resource_path_test.rb
@@ -0,0 +1,58 @@
+# Copyright 2021 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
+#
+#     https://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.
+
+require "helper"
+
+describe Google::Cloud::Firestore::ResourcePath do
+  let(:project_id) { "my-project" }
+  let(:database_id) { "my-database" }
+  let(:doc_id_a) { "alice" }
+  let(:doc_id_b) { "alice-" }
+  let(:doc_id_c) { "bob" }
+  let(:path_a) { "projects/#{project_id}/databases/(default)/documents/users/#{doc_id_a}/pets" }
+  let(:path_b) { "projects/#{project_id}/databases/(default)/documents/users/#{doc_id_b}/pets" }
+  let(:path_c) { "projects/#{project_id}/databases/(default)/documents/users/#{doc_id_c}/pets" }
+  let(:path_d) { "projects/#{project_id}/databases/(default)/documents/users/#{doc_id_c}" }
+  let(:resource_path_a) { Google::Cloud::Firestore::ResourcePath.from_path path_a }
+  let(:resource_path_b) { Google::Cloud::Firestore::ResourcePath.from_path path_b }
+  let(:resource_path_c) { Google::Cloud::Firestore::ResourcePath.from_path path_c }
+  let(:resource_path_d) { Google::Cloud::Firestore::ResourcePath.from_path path_d }
+
+  describe String, "<=>" do
+    it "correctly compares isolated resource IDs" do
+      # "alice" should come before "alice-"
+      _(doc_id_a.<=>(doc_id_b)).must_equal -1
+    end
+    it "incorrectly compares the resource ID, because full paths are sorted dash before slash" do
+      # "alice-" should not come before "alice"
+      _(path_a.<=>(path_b)).must_equal 1
+    end
+  end
+
+  it "parses project_id" do
+    _(resource_path_a.project_id).must_equal project_id
+  end
+
+  it "parses default database_id" do
+    _(resource_path_a.database_id).must_equal "(default)"
+  end
+
+  it "compares a dashed document ID reference" do
+    _(resource_path_a.<=>(resource_path_b)).must_equal -1
+  end
+
+  it "compares a simple document ID reference" do
+    _(resource_path_a.<=>(resource_path_c)).must_equal -1
+  end
+end
diff --git a/google-cloud-firestore/test/helper.rb b/google-cloud-firestore/test/helper.rb
index 56651a21a255..14b169b0f723 100644
--- a/google-cloud-firestore/test/helper.rb
+++ b/google-cloud-firestore/test/helper.rb
@@ -127,7 +127,7 @@ class MockFirestore < Minitest::Spec
   def wait_until &block
     wait_count = 0
     until block.call
-      fail "wait_until criterial was not met" if wait_count > 100
+      fail "wait_until criteria was not met" if wait_count > 100
       wait_count += 1
       sleep 0.01
     end
@@ -182,9 +182,48 @@ def run_query_args query,
     [req, default_options]
   end
 
+  def partition_query_args query_grpc,
+                           parent: "projects/#{project}/databases/(default)/documents",
+                           partition_count: 2,
+                           page_token: nil,
+                           page_size: nil
+    [
+      Google::Cloud::Firestore::V1::PartitionQueryRequest.new(
+        parent: parent,
+        structured_query: query_grpc,
+        partition_count: partition_count,
+        page_token: page_token,
+        page_size: page_size
+      )
+    ]
+  end
+
+  def partition_query_resp doc_ids: ["10", "20"], token: nil
+    Google::Cloud::Firestore::V1::PartitionQueryResponse.new(
+      partitions: doc_ids.map { |id| cursor_grpc doc_ids: [id] },
+      next_page_token: token
+    )
+  end
+
+  def cursor_grpc doc_ids: ["10"], before: true
+    converted_values = doc_ids.map do |doc_id|
+      Google::Cloud::Firestore::V1::Value.new(
+        reference_value: document_path(doc_id)
+      )
+    end
+    Google::Cloud::Firestore::V1::Cursor.new(
+      values: converted_values,
+      before: before
+    )
+  end
+
   def paged_enum_struct response
     OpenStruct.new response: response
   end
+
+  def document_path doc_id
+    "projects/#{project}/databases/(default)/documents/my-collection-id/#{doc_id}"
+  end
 end
 
 class WatchFirestore < MockFirestore