Dm Limits And Best Practices

CDF Data Modeling: Limits, Concurrency & Best Practices

This is a reference skill. When writing or reviewing code that calls CDF Data Modeling APIs, apply the patterns below.

This skill owns runtime reliability concerns: limits, concurrency, retries, throughput, and batching behavior. For traversal payload correctness and graph-specific failure signatures, see dm-graph-traversal.

DMS Limits Reference

For the latest concurrency limits, resource limits, and property value limits, see the official documentation: https://docs.cognite.com/cdf/dm/dm_reference/dm_limits_and_restrictions

Key things to be aware of:

• Instance apply, delete, and query operations each have their own concurrent request limits
• Exceeding these limits returns 429 Too Many Requests
• Transformations consume a large portion of the concurrency budget, leaving less for other clients
• instances.list has a max page size (use pagination for complete results)
• instances.query table expressions each have their own item limit
• instances.upsert accepts up to 1000 items per call
• in filters accept at most 1000 values per expression; larger sets must be split into batches

Search vs Filter: When to Use Which

instances.search — Free-text search on text properties

Use instances.search when you need fuzzy/text matching on string fields (names, descriptions, etc.). It supports an operator parameter:

• `AND` (default) — Narrow search. All terms must match. Use when the user provides a specific query.
• `OR` — Broad "shotgun" search. Any term can match. Use for exploratory/typeahead search where you want maximum recall.

// Narrow search: find a specific cell by name (AND — all terms must match)
const exactResults = await client.instances.search({
  view: { type: 'view', ...PROCESS_CELL_VIEW },
  query: 'reactor tank A',
  properties: ['name'],
  operator: 'AND',
  limit: 10,
});

// Broad search: typeahead/autocomplete (OR — any term can match)
const broadResults = await client.instances.search({
  view: { type: 'view', ...BATCH_VIEW },
  query: 'BUDE completed',
  properties: ['name', 'description', 'batchStatus'],
  operator: 'OR',
  limit: 10,
});

You can combine search with filter to further constrain results with exact-match conditions:

// Text search + exact filter: search for "pump" but only in active nodes
const filtered = await client.instances.search({
  view: { type: 'view', ...PROCESS_CELL_VIEW },
  query: 'pump',
  properties: ['name', 'description'],
  filter: {
    equals: {
      property: getContainerProperty(MY_CONTAINER, 'status'),
      value: 'active',
    },
  },
  limit: 20,
});

instances.list / instances.query with filter — Exact-match filtering

Use filter when you need precise, deterministic matching (equals, range, in, hasData, etc.). No fuzzy matching — values must match exactly.

// Exact match: get all completed batches
const completedBatches = await client.instances.list({
  instanceType: 'node',
  sources: [{ source: { type: 'view', ...BATCH_VIEW } }],
  filter: {
    equals: {
      property: getContainerProperty(BATCH_CONTAINER, 'batchStatus'),
      value: 'completed',
    },
  },
  limit: 1000,
});

Decision Guide

in filter value limit (1000) and batching

CDF in filters support a maximum of 1000 values in a single filter expression. If you need to filter against more than 1000 IDs, split values into chunks and issue multiple requests, then merge results.

const IN_FILTER_BATCH_SIZE = 1000;
// Reuse the Chunking Utility defined in the Batching Write Operations section.

async function listByExternalIds(
  client: CogniteClient,
  externalIds: string[],
): Promise<NodeOrEdge[]> {
  const idBatches = chunk(externalIds, IN_FILTER_BATCH_SIZE);
  const responses = await Promise.all(
    idBatches.map((batch) =>
      cdfTaskRunner.schedule(() =>
        client.instances.list({
          instanceType: 'node',
          sources: [{ source: { type: 'view', ...MY_VIEW } }],
          filter: {
            in: {
              property: ['node', 'externalId'],
              values: batch,
            },
          },
          limit: 1000,
        })
      )
    )
  );

  return responses.flatMap((r) => r.items);
}

QueuedTaskRunner (Semaphore)

Always use the global `cdfTaskRunner` to wrap CDF API calls. It limits concurrent requests and prevents 429 errors and deadlocks.

Source Code

If the project does not already have a semaphore utility, create src/shared/utils/semaphore.ts with this implementation:

/**
 * AbortError thrown when a queued task is cancelled
 */
export class AbortError extends Error {
  public constructor(message: string = 'Aborted') {
    super(message);
    this.name = 'AbortError';
  }
}

type PendingTask<AsyncFn, AsyncFnResult> = {
  resolve: (result: AsyncFnResult) => void;
  reject: (error: unknown) => void;
  fn: AsyncFn;
  key?: string;
};

const DEFAULT_MAX_CONCURRENT_TASKS = 15;

/**
 * QueuedTaskRunner for controlling concurrent operations
 * Used to limit concurrent CDF API requests to avoid rate limiting and deadlocks
 * Essentially a semaphore that allows a limited number of tasks to run at once.
 */
export default class QueuedTaskRunner<
  AsyncFn extends () => Promise<AsyncFnResult>,
  AsyncFnResult = Awaited<ReturnType<AsyncFn>>,
> {
  private pendingTasks: PendingTask<AsyncFn, AsyncFnResult>[] = [];
  private currentPendingTasks: number = 0;
  private readonly maxConcurrentTasks: number = 1;

  public constructor(
    maxConcurrentTasks: number = DEFAULT_MAX_CONCURRENT_TASKS
  ) {
    this.maxConcurrentTasks = maxConcurrentTasks;
  }

  public schedule(
    fn: AsyncFn,
    options: { key?: string } = {}
  ): Promise<AsyncFnResult> {
    this.startTrackingTime();

    return new Promise((resolve, reject) => {
      if (options.key !== undefined) {
        // Cancel existing tasks with the same key (deduplication)
        this.pendingTasks
          .filter((task) => task.key === options.key)
          .forEach((task) => task.reject(new AbortError()));

        this.pendingTasks = this.pendingTasks.filter(
          (task) => task.key !== options.key
        );
      }

      this.pendingTasks.push({
        resolve,
        reject,
        fn,
        key: options.key,
      });

      this.attemptConsumingNextTask();
    });
  }

  public async attemptConsumingNextTask(): Promise<void> {
    if (this.pendingTasks.length === 0) return;
    if (this.currentPendingTasks >= this.maxConcurrentTasks) return;

    const pendingTask = this.pendingTasks.shift();
    if (pendingTask === undefined) {
      throw new Error('pendingTask is undefined, this should never happen');
    }

    this.currentPendingTasks++;
    const { fn, resolve, reject } = pendingTask;

    try {
      const result = await fn();
      resolve(result);
    } catch (e) {
      reject(e);
    } finally {
      this.currentPendingTasks--;
      this.tick();
      this.attemptConsumingNextTask();
    }
  }

  public clearQueue = (): void => {
    this.pendingTasks = [];
  };

  private startTime: number | null = null;

  private startTrackingTime = (): void => {
    if (this.startTime === null) {
      this.startTime = performance.now();
    }
  };

  private tick = (): void => {
    if (this.pendingTasks.length === 0) {
      this.startTime = null;
    }
  };
}

/**
 * Global task runner for CDF API requests