diff --git a/src/blob/handlers/ContainerHandler.ts b/src/blob/handlers/ContainerHandler.ts
index 66c40af6d..e2d225039 100644
--- a/src/blob/handlers/ContainerHandler.ts
+++ b/src/blob/handlers/ContainerHandler.ts
@@ -680,7 +680,8 @@ export default class ContainerHandler extends BaseHandler
       options.maxresults,
       marker,
       includeSnapshots,
-      includeUncommittedBlobs
+      includeUncommittedBlobs,
+      request.getQuery("startFrom")
     );
 
     const serviceEndpoint = `${request.getEndpoint()}/${accountName}`;
@@ -786,7 +787,8 @@ export default class ContainerHandler extends BaseHandler
       options.maxresults,
       marker,
       includeSnapshots,
-      includeUncommittedBlobs
+      includeUncommittedBlobs,
+      request.getQuery("startFrom")
     );
 
     const serviceEndpoint = `${request.getEndpoint()}/${accountName}`;
diff --git a/src/blob/persistence/IBlobMetadataStore.ts b/src/blob/persistence/IBlobMetadataStore.ts
index fb933f8df..fc492c8b9 100644
--- a/src/blob/persistence/IBlobMetadataStore.ts
+++ b/src/blob/persistence/IBlobMetadataStore.ts
@@ -495,7 +495,8 @@ export interface IBlobMetadataStore
     maxResults?: number,
     marker?: string,
     includeSnapshots?: boolean,
-    includeUncommittedBlobs?: boolean
+    includeUncommittedBlobs?: boolean,
+    startFrom?: string
   ): Promise<[BlobModel[], BlobPrefixModel[], string | undefined]>;
 
   listAllBlobs(
diff --git a/src/blob/persistence/LokiBlobMetadataStore.ts b/src/blob/persistence/LokiBlobMetadataStore.ts
index d0e3f62d6..8dc81f1a4 100644
--- a/src/blob/persistence/LokiBlobMetadataStore.ts
+++ b/src/blob/persistence/LokiBlobMetadataStore.ts
@@ -919,7 +919,8 @@ export default class LokiBlobMetadataStore
     maxResults: number = DEFAULT_LIST_BLOBS_MAX_RESULTS,
     marker: string = "",
     includeSnapshots?: boolean,
-    includeUncommittedBlobs?: boolean
+    includeUncommittedBlobs?: boolean,
+    startFrom?: string
   ): Promise<[BlobModel[], BlobPrefixModel[], string | undefined]> {
     const query: any = {};
     if (prefix !== "") {
@@ -948,6 +949,12 @@ export default class LokiBlobMetadataStore
         .where((obj) => {
           return obj.name > marker!;
         })
+        .where((obj) => {
+          // startFrom is inclusive where marker is exclusive, and the two
+          // compose: paging a listing that began at startFrom advances the
+          // marker past it anyway.
+          return startFrom === undefined ? true : obj.name >= startFrom;
+        })
         .where((obj) => {
           return includeSnapshots ? true : obj.snapshot.length === 0;
         })
diff --git a/src/blob/persistence/SqlBlobMetadataStore.ts b/src/blob/persistence/SqlBlobMetadataStore.ts
index 8443360b7..d5723e02d 100644
--- a/src/blob/persistence/SqlBlobMetadataStore.ts
+++ b/src/blob/persistence/SqlBlobMetadataStore.ts
@@ -1300,7 +1300,8 @@ export default class SqlBlobMetadataStore implements IBlobMetadataStore {
     maxResults: number = DEFAULT_LIST_BLOBS_MAX_RESULTS,
     marker?: string,
     includeSnapshots?: boolean,
-    includeUncommittedBlobs?: boolean
+    includeUncommittedBlobs?: boolean,
+    startFrom?: string
   ): Promise<[BlobModel[], BlobPrefixModel[], any | undefined]> {
     return this.sequelize.transaction(async (t) => {
       await this.assertContainerExists(context, account, container, t);
@@ -1328,6 +1329,19 @@ export default class SqlBlobMetadataStore implements IBlobMetadataStore {
             };
           }
         }
+
+        // startFrom is inclusive where marker is exclusive, and the two
+        // compose: paging a listing that began at startFrom advances the
+        // marker past it anyway.
+        if (startFrom !== undefined) {
+          if (whereQuery.blobName !== undefined) {
+            whereQuery.blobName[Op.gte] = startFrom;
+          } else {
+            whereQuery.blobName = {
+              [Op.gte]: startFrom
+            };
+          }
+        }
       }
       if (!includeSnapshots) {
         whereQuery.snapshot = "";
diff --git a/tests/blob/apis/container.test.ts b/tests/blob/apis/container.test.ts
index 044fcad30..07f99a2b7 100644
--- a/tests/blob/apis/container.test.ts
+++ b/tests/blob/apis/container.test.ts
@@ -4,6 +4,7 @@ import {
   AccountSASServices,
   AnonymousCredential,
   BlobServiceClient,
+  ContainerClient,
   generateAccountSASQueryParameters,
   newPipeline,
   StorageSharedKeyCredential,
@@ -1712,4 +1713,100 @@ describe("ContainerAPIs", () => {
     assert.strictEqual(false, await containerClient.exists());
     assert.strictEqual(false, await blockBlobClient.exists());
   });
+
+  it("listBlobsFlat with startFrom should begin at that blob name @loki @sql", async () => {
+    // "b&c" needs encoding to survive the query string: a bare & ends the
+    // parameter, leaving a startFrom of "b" that admits one blob too many.
+    const names = ["a", "b", "b&c", "c"];
+    for (const name of names) {
+      await containerClient.getBlockBlobClient(name).upload("", 0);
+    }
+
+    const listFrom = async (startFrom: string) => {
+      const page = (
+        await listBlobsWithStartFrom(startFrom).listBlobsFlat().byPage().next()
+      ).value;
+      return page.segment.blobItems.map((item: any) => item.name);
+    };
+
+    // startFrom is inclusive, unlike the marker
+    assert.deepStrictEqual(await listFrom("b"), ["b", "b&c", "c"]);
+    assert.deepStrictEqual(await listFrom("b&c"), ["b&c", "c"]);
+    // A name no blob carries still starts the listing at the right place
+    assert.deepStrictEqual(await listFrom("b0"), ["c"]);
+    // One before every blob returns them all, as delta lake asks for
+    assert.deepStrictEqual(await listFrom("0"), names);
+
+    for (const name of names) {
+      await containerClient.getBlockBlobClient(name).delete();
+    }
+  });
+
+  it("listBlobsByHierarchy with startFrom should begin at that blob name @loki @sql", async () => {
+    const names = ["a/1", "a/2", "b/1", "c"];
+    for (const name of names) {
+      await containerClient.getBlockBlobClient(name).upload("", 0);
+    }
+
+    const listFrom = async (startFrom: string) => {
+      const page = (
+        await listBlobsWithStartFrom(startFrom)
+          .listBlobsByHierarchy("/")
+          .byPage()
+          .next()
+      ).value;
+      return [
+        (page.segment.blobPrefixes ?? []).map((p: any) => p.name),
+        page.segment.blobItems.map((item: any) => item.name)
+      ];
+    };
+
+    // startFrom selects blobs before the delimiter squashes them into
+    // prefixes, so a prefix survives only while one of its blobs does.
+    assert.deepStrictEqual(await listFrom("0"), [["a/", "b/"], ["c"]]);
+    assert.deepStrictEqual(await listFrom("a/2"), [["a/", "b/"], ["c"]]);
+    assert.deepStrictEqual(await listFrom("b/"), [["b/"], ["c"]]);
+    assert.deepStrictEqual(await listFrom("c"), [[], ["c"]]);
+
+    for (const name of names) {
+      await containerClient.getBlockBlobClient(name).delete();
+    }
+  });
+
+  // The JS SDK has no startFrom yet, so put it on the wire by rewriting the
+  // query of a SAS-authenticated request.  A blob name is arbitrary text, so
+  // it is encoded the way a client would have to encode it.
+  function listBlobsWithStartFrom(startFrom: string): ContainerClient {
+    const storageSharedKeyCredential = new StorageSharedKeyCredential(
+      EMULATOR_ACCOUNT_NAME,
+      EMULATOR_ACCOUNT_KEY
+    );
+    const tmr = new Date();
+    tmr.setDate(tmr.getDate() + 1);
+    const sas = generateAccountSASQueryParameters(
+      {
+        expiresOn: tmr,
+        permissions: AccountSASPermissions.parse("rl"),
+        resourceTypes: AccountSASResourceTypes.parse("sco").toString(),
+        services: AccountSASServices.parse("b").toString(),
+        version: "2020-04-08"
+      },
+      storageSharedKeyCredential as StorageSharedKeyCredential
+    ).toString();
+
+    const pipeline = newPipeline(new AnonymousCredential(), {
+      retryOptions: { maxTries: 1 },
+      keepAliveOptions: { enable: false }
+    });
+    pipeline.factories.unshift(
+      new QueryRequestPolicyFactory(
+        "restype=container",
+        `restype=container&startFrom=${encodeURIComponent(startFrom)}`
+      )
+    );
+    return new BlobServiceClient(
+      `${baseURL}?${sas}`,
+      pipeline
+    ).getContainerClient(containerName);
+  }
 });