Add support for LIMIT and OFFSET in queries (Azure#306)

src/queryExecutionContext/EndpointComponent/AggregateEndpointComponent.ts

@@ -1,12 +1,11 @@
 import { Response } from "../../request";
 import { AverageAggregator, CountAggregator, MaxAggregator, MinAggregator, SumAggregator } from "../Aggregators";
+import { ExecutionContext } from "../ExecutionContext";
 import { getInitialHeader } from "../headerUtils";
-import { IExecutionContext } from "../IExecutionContext";
 import { CosmosHeaders } from "../index";
-import { IEndpointComponent } from "./IEndpointComponent";
 
 /** @hidden */
-export class AggregateEndpointComponent implements IEndpointComponent {
+export class AggregateEndpointComponent implements ExecutionContext {
   private toArrayTempResources: any[];
   private aggregateValues: any[];
   private aggregateValuesIndex: number;
@@ -18,7 +17,7 @@ export class AggregateEndpointComponent implements IEndpointComponent {
    * @param { object } executionContext - Underlying Execution Context
    * @ignore
    */
-  constructor(private executionContext: IExecutionContext, aggregateOperators: string[]) {
+  constructor(private executionContext: ExecutionContext, aggregateOperators: string[]) {
     // TODO: any
     this.executionContext = executionContext;
     this.localAggregators = [];

src/queryExecutionContext/EndpointComponent/OffsetLimitEndpointComponent.ts

@@ -0,0 +1,36 @@
+import { Response } from "../../request";
+import { ExecutionContext } from "../ExecutionContext";
+import { getInitialHeader } from "../headerUtils";
+
+/** @hidden */
+export class OffsetLimitEndpointComponent implements ExecutionContext {
+  constructor(private executionContext: ExecutionContext, private offset: number, private limit: number) {}
+
+  public async nextItem(): Promise<Response<any>> {
+    if (this.offset > 0) {
+      // Grab next item but ignore the result. We only need the headers
+      const { headers } = await this.executionContext.nextItem();
+      this.offset--;
+      return { result: undefined, headers };
+    }
+    if (this.limit > 0) {
+      const response = await this.executionContext.nextItem();
+      this.limit--;
+      return response;
+    }
+    // If both limit and offset are 0, return nothing
+    return { result: undefined, headers: getInitialHeader() };
+  }
+
+  public async current(): Promise<Response<any>> {
+    if (this.offset > 0) {
+      const current = await this.executionContext.current();
+      return { result: undefined, headers: current.headers };
+    }
+    return this.executionContext.current();
+  }
+
+  public hasMoreResults() {
+    return (this.offset > 0 || this.limit > 0) && this.executionContext.hasMoreResults();
+  }
+}

src/queryExecutionContext/EndpointComponent/OrderByEndpointComponent.ts

@@ -1,53 +1,40 @@
 import { Response } from "../../request";
-import { IExecutionContext } from "../IExecutionContext";
-import { IEndpointComponent } from "./IEndpointComponent";
+import { ExecutionContext } from "../ExecutionContext";
 
 /** @hidden */
-export class OrderByEndpointComponent implements IEndpointComponent {
+export class OrderByEndpointComponent implements ExecutionContext {
   /**
    * Represents an endpoint in handling an order by query. For each processed orderby \
    * result it returns 'payload' item of the result
    * @constructor OrderByEndpointComponent
    * @param {object} executionContext              - Underlying Execution Context
    * @ignore
    */
-  constructor(private executionContext: IExecutionContext) {}
+  constructor(private executionContext: ExecutionContext) {}
   /**
    * Execute a provided function on the next element in the OrderByEndpointComponent.
    * @memberof OrderByEndpointComponent
    * @instance
-   * @param {callback} callback - Function to execute for each element. the function \
-   * takes two parameters error, element.
    */
   public async nextItem(): Promise<Response<any>> {
-    try {
-      const { result: item, headers } = await this.executionContext.nextItem();
-      return {
-        result: item !== undefined ? item.payload : undefined,
-        headers
-      };
-    } catch (err) {
-      throw err;
-    }
+    const { result: item, headers } = await this.executionContext.nextItem();
+    return {
+      result: item !== undefined ? item.payload : undefined,
+      headers
+    };
   }
 
   /**
    * Retrieve the current element on the OrderByEndpointComponent.
    * @memberof OrderByEndpointComponent
    * @instance
-   * @param {callback} callback - Function to execute for the current element. \
-   * the function takes two parameters error, element.
    */
   public async current(): Promise<Response<any>> {
-    try {
-      const { result: item, headers } = await this.executionContext.current();
-      return {
-        result: item !== undefined ? item.payload : undefined,
-        headers
-      };
-    } catch (err) {
-      throw err;
-    }
+    const { result: item, headers } = await this.executionContext.current();
+    return {
+      result: item !== undefined ? item.payload : undefined,
+      headers
+    };
   }
 
   /**

src/queryExecutionContext/IExecutionContext.ts → src/queryExecutionContext/ExecutionContext.ts

@@ -1,7 +1,7 @@
 import { Response } from "../request";
 
 /** @hidden */
-export interface IExecutionContext {
+export interface ExecutionContext {
   nextItem: () => Promise<Response<any>>;
   current: () => Promise<Response<any>>;
   hasMoreResults: () => boolean;

src/queryExecutionContext/defaultQueryExecutionContext.ts

@@ -2,7 +2,7 @@ import { Constants } from "../common";
 import { ClientSideMetrics, QueryMetrics } from "../queryMetrics";
 import { Response } from "../request";
 import { getInitialHeader } from "./headerUtils";
-import { IExecutionContext } from "./index";
+import { ExecutionContext } from "./index";
 
 /** @hidden */
 export type FetchFunctionCallback = (options: any) => Promise<Response<any>>;
@@ -15,7 +15,7 @@ enum STATES {
 }
 
 /** @hidden */
-export class DefaultQueryExecutionContext implements IExecutionContext {
+export class DefaultQueryExecutionContext implements ExecutionContext {
   private static readonly STATES = STATES;
   private resources: any; // TODO: any resources
   private currentIndex: number;

src/queryExecutionContext/index.ts

@@ -2,11 +2,10 @@ export * from "./headerUtils";
 export * from "./SqlQuerySpec";
 export * from "./defaultQueryExecutionContext";
 export * from "./Aggregators";
-export * from "./EndpointComponent";
 export * from "./documentProducer";
 export * from "./FetchResult";
 export * from "./orderByDocumentProducerComparator";
-export * from "./IExecutionContext";
+export * from "./ExecutionContext";
 export * from "./parallelQueryExecutionContextBase";
 export * from "./parallelQueryExecutionContext";
 export * from "./orderByQueryExecutionContext";

src/queryExecutionContext/orderByQueryExecutionContext.ts

@@ -1,12 +1,12 @@
 import { ClientContext } from "../ClientContext";
 import { PartitionedQueryExecutionInfo } from "../request/ErrorResponse";
 import { DocumentProducer } from "./documentProducer";
-import { IExecutionContext } from "./IExecutionContext";
+import { ExecutionContext } from "./ExecutionContext";
 import { OrderByDocumentProducerComparator } from "./orderByDocumentProducerComparator";
 import { ParallelQueryExecutionContextBase } from "./parallelQueryExecutionContextBase";
 
 /** @hidden */
-export class OrderByQueryExecutionContext extends ParallelQueryExecutionContextBase implements IExecutionContext {
+export class OrderByQueryExecutionContext extends ParallelQueryExecutionContextBase implements ExecutionContext {
   private orderByComparator: any;
   /**
    * Provides the OrderByQueryExecutionContext.

src/queryExecutionContext/parallelQueryExecutionContext.ts

@@ -1,11 +1,11 @@
 import { ClientContext } from "../ClientContext";
 import { PartitionedQueryExecutionInfo } from "../request/ErrorResponse";
 import { DocumentProducer } from "./documentProducer";
-import { IExecutionContext } from "./IExecutionContext";
+import { ExecutionContext } from "./ExecutionContext";
 import { ParallelQueryExecutionContextBase } from "./parallelQueryExecutionContextBase";
 
 /** @hidden */
-export class ParallelQueryExecutionContext extends ParallelQueryExecutionContextBase implements IExecutionContext {
+export class ParallelQueryExecutionContext extends ParallelQueryExecutionContextBase implements ExecutionContext {
   /**
    * Provides the ParallelQueryExecutionContext.
    * This class is capable of handling parallelized queries and dervives from ParallelQueryExecutionContextBase.

src/queryExecutionContext/parallelQueryExecutionContextBase.ts

@@ -9,8 +9,8 @@ import { QueryRange } from "../routing/QueryRange";
 import { PARITIONKEYRANGE, SmartRoutingMapProvider } from "../routing/smartRoutingMapProvider";
 import { CosmosHeaders } from "./CosmosHeaders";
 import { DocumentProducer } from "./documentProducer";
+import { ExecutionContext } from "./ExecutionContext";
 import { getInitialHeader, mergeHeaders } from "./headerUtils";
-import { IExecutionContext } from "./IExecutionContext";
 
 /** @hidden */
 export enum ParallelQueryExecutionContextBaseStates {
@@ -20,7 +20,7 @@ export enum ParallelQueryExecutionContextBaseStates {
 }
 
 /** @hidden */
-export abstract class ParallelQueryExecutionContextBase implements IExecutionContext {
+export abstract class ParallelQueryExecutionContextBase implements ExecutionContext {
   private static readonly DEFAULT_PAGE_SIZE = 10;
 
   private err: any;

src/queryExecutionContext/pipelinedQueryExecutionContext.ts

@@ -2,22 +2,19 @@ import { ClientContext } from "../ClientContext";
 import { Response } from "../request";
 import { PartitionedQueryExecutionInfo } from "../request/ErrorResponse";
 import { CosmosHeaders } from "./CosmosHeaders";
-import {
-  AggregateEndpointComponent,
-  IEndpointComponent,
-  OrderByEndpointComponent,
-  TopEndpointComponent
-} from "./EndpointComponent";
+import { AggregateEndpointComponent } from "./EndpointComponent/AggregateEndpointComponent";
+import { OffsetLimitEndpointComponent } from "./EndpointComponent/OffsetLimitEndpointComponent";
+import { OrderByEndpointComponent } from "./EndpointComponent/OrderByEndpointComponent";
+import { ExecutionContext } from "./ExecutionContext";
 import { getInitialHeader, mergeHeaders } from "./headerUtils";
-import { IExecutionContext } from "./IExecutionContext";
 import { OrderByQueryExecutionContext } from "./orderByQueryExecutionContext";
 import { ParallelQueryExecutionContext } from "./parallelQueryExecutionContext";
 
 /** @hidden */
-export class PipelinedQueryExecutionContext implements IExecutionContext {
+export class PipelinedQueryExecutionContext implements ExecutionContext {
   private fetchBuffer: any[];
   private fetchMoreRespHeaders: CosmosHeaders;
-  private endpoint: IEndpointComponent;
+  private endpoint: ExecutionContext;
   private pageSize: number;
   private static DEFAULT_PAGE_SIZE = 10;
   constructor(
@@ -63,10 +60,17 @@ export class PipelinedQueryExecutionContext implements IExecutionContext {
       this.endpoint = new AggregateEndpointComponent(this.endpoint, aggregates);
     }
 
-    // If top then add that to the pipeline
+    // If top then add that to the pipeline. TOP N is effectively OFFSET 0 LIMIT N
     const top = partitionedQueryExecutionInfo.queryInfo.top;
     if (typeof top === "number") {
-      this.endpoint = new TopEndpointComponent(this.endpoint, top);
+      this.endpoint = new OffsetLimitEndpointComponent(this.endpoint, 0, top);
+    }
+
+    // If offset+limit then add that to the pipeline
+    const limit = partitionedQueryExecutionInfo.queryInfo.limit;
+    const offset = partitionedQueryExecutionInfo.queryInfo.offset;
+    if (typeof limit === "number" && typeof offset === "number") {
+      this.endpoint = new OffsetLimitEndpointComponent(this.endpoint, offset, limit);
     }
   }
 

src/queryExecutionContext/proxyQueryExecutionContext.ts

@@ -3,13 +3,13 @@ import { StatusCodes, SubStatusCodes } from "../common/statusCodes";
 import { ErrorResponse, Response } from "../request";
 import { PartitionedQueryExecutionInfo } from "../request/ErrorResponse";
 import { DefaultQueryExecutionContext, FetchFunctionCallback } from "./defaultQueryExecutionContext";
-import { IExecutionContext } from "./IExecutionContext";
+import { ExecutionContext } from "./ExecutionContext";
 import { PipelinedQueryExecutionContext } from "./pipelinedQueryExecutionContext";
 import { SqlQuerySpec } from "./SqlQuerySpec";
 
 /** @hidden */
-export class ProxyQueryExecutionContext implements IExecutionContext {
-  private queryExecutionContext: IExecutionContext;
+export class ProxyQueryExecutionContext implements ExecutionContext {
+  private queryExecutionContext: ExecutionContext;
 
   constructor(
     private clientContext: ClientContext,