dm-limits-and-best-practices

CDF データモデリング：制限、並行処理、ベストプラクティス

これはリファレンススキルです。CDF Data Modeling API を呼び出すコードを作成またはレビューする際は、以下のパターンを適用してください。

このスキルでは、制限、並行処理、リトライ、スループット、バッチ処理の動作といった、ランタイムの信頼性に関する事項を扱います。 トラバーサルのペイロードの正確性とグラフ固有の失敗パターンについては、dm-graph-traversal を参照してください。

---

DMS の制限に関するリファレンス

最新の並行処理制限、リソース制限、プロパティ値の制限については、公式ドキュメントを参照してください： https://docs.cognite.com/cdf/dm/dm_reference/dm_limits_and_restrictions

把握しておくべき主な点：

• インスタンスの apply、delete、query 操作には、それぞれ独自の同時リクエスト制限があります
• これらの制限を超えると、429 Too Many Requests が返されます
• Transformations は並行処理の予算を大きく消費するため、ほかのクライアントが利用できる枠が少なくなります
• instances.list には最大ページサイズがあります（結果をすべて取得するにはページネーションを使用してください）
• instances.query の各テーブル式には、それぞれ独自の項目数制限があります
• instances.upsert は、1 回の呼び出しで最大 1000 項目を受け付けます
• in フィルターは式あたり最大 1000 個の値を受け付けます。それより大きなセットはバッチに分割する必要があります

---

Search と Filter：使い分け

instances.search — テキストプロパティの全文検索

文字列フィールド（名前、説明など）であいまい検索やテキストマッチングが必要な場合は、instances.search を使用してください。operator パラメーターをサポートしています：

• AND（デフォルト）— 範囲を絞り込む検索です。すべての語句が一致する必要があります。ユーザーが具体的なクエリを指定した場合に使用します。
• OR — 範囲を広げる「手当たり次第」の検索です。いずれかの語句が一致すればよい場合に使用します。再現率を最大化したい探索的な検索や入力補完検索に適しています。

// 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,
});

search と filter を組み合わせると、完全一致条件で結果をさらに絞り込めます：

// 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,
});

filter を使った instances.list / instances.query — 完全一致フィルタリング

正確で決定的なマッチング（等価、範囲、in、hasData など）が必要な場合は、filter を使用してください。あいまい検索は行われず、値は完全に一致する必要があります。

// 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,
});

選択ガイド

必要なこと	使用するもの
検索ボックスへのユーザー入力	OR を使った instances.search
名前で特定の項目を見つける	AND を使った instances.search
ステータス、日付範囲、列挙値で絞り込む	list/query の filter
テキスト検索と完全一致条件の併用	instances.search + filter

in フィルターの値の上限（1000）とバッチ処理

CDF の in フィルターでは、1 つのフィルター式に最大 1000 個の値を指定できます。1000 個を超える ID を対象にフィルタリングする必要がある場合は、値をチャンクに分割して複数のリクエストを発行し、その後、結果を統合してください。

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（セマフォ）

CDF API 呼び出しをラップするには、必ずグローバルな cdfTaskRunner を使用してください。同時リクエスト数を制限し、429 エラーやデッドロックを防ぎます。

ソースコード

プロジェクトにセマフォユーティリティがまだない場合は、この実装を使って src/shared/utils/semaphore.ts を作成してください：

/**
 * 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
 * Limits concurrent requests to avoid 429 rate limiting and deadlocks
 */
export const cdfTaskRunner = new QueuedTaskRunner(DEFAULT_MAX_CONCURRENT_TASKS);

使用パターン

CDF の呼び出しは必ず cdfTaskRunner.schedule() でラップしてください：

import { cdfTaskRunner } from '../../../../shared/utils/semaphore';

// Single query
export async function fetchBatches(client: CogniteClient): Promise<CDFBatch[]> {
  return cdfTaskRunner.schedule(async () => {
    const response = await client.instances.query({
      with: { /* ... */ },
      select: { /* ... */ },
    });
    return response.items?.nodes || [];
  });
}

// Multiple parallel queries (safe — the semaphore limits concurrency)
export async function enrichBatch(
  client: CogniteClient,
  batch: CDFBatch
): Promise<BatchEnrichment> {
  const [currentOp, lastOp, cells, material] = await Promise.all([
    fetchCurrentOperation(client, batch.space, batch.externalId),
    fetchLastCompletedOperation(client, batch.space, batch.externalId),
    fetchProcessCells(client, batch.space, batch.externalId),
    fetchMaterial(client, batch.space, batch.externalId),
  ]);
  return { currentOp, lastOp, cells, material };
}

// Each of the above functions internally uses cdfTaskRunner.schedule(),
// so Promise.all is safe — the semaphore prevents exceeding concurrency limits

キーを使った重複排除

同じクエリが再度実行されたときに古いリクエストをキャンセルするには、key オプションを使用してください（例：ユーザーがフィルターを素早く変更する場合）：

const result = await cdfTaskRunner.schedule(
  async () => client.instances.query({ /* ... */ }),
  { key: `batch-flow-${batchId}` }
);
// If another call with the same key arrives before this completes,
// the previous pending call is rejected with AbortError

---

ページネーション

DMS の instances.list は、最大で limit 個の項目と、次のページ用の nextCursor を返します。 DMS の instances.query は、テーブル式名をキーとする cursors オブジェクトを使用します。

instances.list のページネーション

async function fetchAllNodes(client: CogniteClient): Promise<CDFNodeResponse[]> {
  const allItems: CDFNodeResponse[] = [];
  let cursor: string | undefined = undefined;

  do {
    const response = await client.instances.list({
      instanceType: 'node',
      sources: [{ source: { type: 'view', ...MY_VIEW } }],
      filter: {
        equals: {
          property: getContainerProperty(MY_CONTAINER, 'status'),
          value: 'active',
        },
      },
      limit: 1000,
      cursor,
    });

    allItems.push(...response.items);
    cursor = response.nextCursor;
  } while (cursor);

  return allItems;
}

instances.query のページネーション

query エンドポイントは、nextCursor を Record<string, string>（テーブル式ごとに 1 つのカーソル）として返します。これを cursors パラメーターで使用してください：

import { isEmpty } from 'lodash';

async function fetchAllResults(
  client: CogniteClient
): Promise<{ results: CDFResult[]; edges: EdgeDefinition[] }> {
  const QUERY_LIMIT = 10_000;

  const fetchPage = async (
    nextCursors?: Record<string, string>
  ): Promise<{ results: CDFResult[]; edges: EdgeDefinition[] }> => {
    const { items, nextCursor } = await client.instances.query({
      with: {
        results: {
          limit: QUERY_LIMIT,
          nodes: {
            filter: {
              hasData: [{ type: 'view', ...RESULT_VIEW }],
            },
          },
        },
        relatedEdges: {
          limit: QUERY_LIMIT,
          edges: {
            from: 'results' as const,
            maxDistance: 1,
            direction: 'outwards' as const,
            filter: {
              equals: {
                property: ['edge', 'type'],
                value: MY_EDGE_TYPE,
              },
            },
          },
        },
      },
      cursors: nextCursors, // Pass cursors from previous page
      select: {
        results: {
          sources: [
            { source: { type: 'view', ...RESULT_VIEW }, properties: ['*'] },
          ],
        },
        relatedEdges: {},
      },
    });

    const results = (items?.results || []) as CDFResult[];
    const edges = (items?.relatedEdges || []).filter(
      (e) => e.instanceType === 'edge'
    );

    // Recurse if more pages exist
    if (!isEmpty(nextCursor)) {
      const next = await fetchPage(nextCursor);
      return {
        results: [...results, ...next.results],
        edges: [...edges, ...next.edges],
      };
    }

    return { results, edges };
  };

  return fetchPage();
}

ページネーションと QueuedTaskRunner の併用

並行処理の予算を使い切らないよう、ページネーションを使ったデータ取得は必ずセマフォでラップしてください：

export async function fetchAllWithPagination(
  client: CogniteClient
): Promise<CDFNodeResponse[]> {
  return cdfTaskRunner.schedule(async () => {
    const allItems: CDFNodeResponse[] = [];
    let cursor: string | undefined = undefined;

    do {
      const response = await client.instances.list({
        instanceType: 'node',
        sources: [{ source: { type: 'view', ...MY_VIEW } }],
        filter: { /* ... */ },
        limit: 1000,
        cursor,
      });

      allItems.push(...response.items);
      cursor = response.nextCursor;

      // Optional: break early if you have enough data
      if (allItems.length >= 500) break;
    } while (cursor);

    return allItems;
  });
}

---

書き込み操作のバッチ処理

多数のインスタンスをアップサートする場合は、適用時の同時実行数の上限を超えないよう、チャンクに分割してください。各 instances.upsert 呼び出しで受け付けられる項目は最大 1000 件です。

チャンク分割ユーティリティ

function chunk<T>(arr: T[], size: number): T[][] {
  const chunks: T[][] = [];
  for (let i = 0; i < arr.length; i += size) {
    chunks.push(arr.slice(i, i + size));
  }
  return chunks;
}

QueuedTaskRunner を使用したバッチアップサート

const UPSERT_BATCH_SIZE = 1000;

async function batchUpsertNodes(
  client: CogniteClient,
  nodes: NodeOrEdgeCreate[]
): Promise<void> {
  const chunks = chunk(nodes, UPSERT_BATCH_SIZE);

  // Process chunks through the semaphore — safe even with Promise.all
  await Promise.all(
    chunks.map((batch) =>
      cdfTaskRunner.schedule(async () => {
        await client.instances.upsert({
          items: batch,
        });
      })
    )
  );
}

QueuedTaskRunner を使用したバッチ削除

インスタンスの削除には、さらに厳しい同時実行数の上限があります。別途、より制限の厳しいタスクランナーを使用してください。

import QueuedTaskRunner from '../../../../shared/utils/semaphore';

// Dedicated runner for deletes (stricter concurrency — check docs for current limit)
const deleteTaskRunner = new QueuedTaskRunner(2);

async function batchDeleteNodes(
  client: CogniteClient,
  nodeIds: { space: string; externalId: string }[]
): Promise<void> {
  const chunks = chunk(nodeIds, 1000);

  for (const batch of chunks) {
    await deleteTaskRunner.schedule(async () => {
      await client.instances.delete(
        batch.map((id) => ({
          instanceType: 'node' as const,
          ...id,
        }))
      );
    });
  }
}

---

ハードゲート — クエリ結果に対する LLM 呼び出し

instances.list／query／search のヒットを対象に、チャット補完をマッピングして実行しないでください。Atlas／EOS のサイドバーで 1 回のターン（integrate-fusion-agent）を使用する方法を優先してください。項目ごとの補完が必要な場合は、ユーザーの操作 1 回につき 5 件、上限 50 件とし、space:externalId:lastUpdatedTime をキーにキャッシュし、ユーザーが開始した場合に限って実行してください。

---

よくある落とし穴

1. セマフォ呼び出しのネストによるデッドロック

関数 A がセマフォのスロットを保持したまま、同じくスロットを必要とする関数 B を呼び出すと、すべてのスロットが使用中の場合にデッドロックが発生する可能性があります。セマフォは最も外側の呼び出しレベルで使用するか、内側の呼び出しで同じセマフォを経由しないようにしてください。

// BAD: Nested semaphore — can deadlock
async function fetchAndEnrich(client: CogniteClient) {
  return cdfTaskRunner.schedule(async () => {
    const batches = await fetchBatches(client); // This also calls cdfTaskRunner.schedule!
    // If all slots are held by fetchAndEnrich callers, fetchBatches will never run
  });
}

// GOOD: Let inner functions own the semaphore
async function fetchAndEnrich(client: CogniteClient) {
  const batches = await fetchBatches(client); // Has its own semaphore call
  const enriched = await Promise.all(
    batches.map((b) => enrichBatch(client, b)) // Each has its own semaphore call
  );
  return enriched;
}

2. ページネーションを忘れる

DMS が返す項目数は最大でも limit 件です。ページネーションを行わないと、データが気付かないうちに失われます。必ず nextCursor を確認してください。

// BAD: May miss data
const response = await client.instances.list({ limit: 1000, /* ... */ });
const items = response.items; // Could be incomplete!

// GOOD: Paginate
const allItems = [];
let cursor;
do {
  const response = await client.instances.list({ limit: 1000, cursor, /* ... */ });
  allItems.push(...response.items);
  cursor = response.nextCursor;
} while (cursor);

3. セマフォなしでの無制限な Promise.all

多数の API 呼び出しを並列に実行すると、すぐに 429 の上限に達します。

// BAD: Too many simultaneous requests
await Promise.all(batchIds.map((id) => client.instances.query({ /* ... */ })));

// GOOD: Each call goes through the semaphore
await Promise.all(
  batchIds.map((id) =>
    cdfTaskRunner.schedule(() => client.instances.query({ /* ... */ }))
  )
);

4. テーブル式ごとのクエリ上限

instances.query の各テーブル式には、それぞれ独自の limit があります。1 つの式でのトラバーサル結果がこの上限を超える可能性がある場合は、cursors パラメーターを使用してページネーションする必要があります。

5. サイズの大きすぎる in フィルター

in フィルターでは、式ごとに最大 1000 個の値を指定できます。1 つの in フィルターに 1000 個を超える値を渡すと、エンドポイントやバージョンによっては失敗したり、動作が不完全になったりする可能性があります。必ず値をチャンクに分割して、バッチリクエストを実行してください。

---

確認事項のまとめ

• すべての CDF API 呼び出しを cdfTaskRunner.schedule() でラップする
• instances.list 呼び出しを cursor／nextCursor を使用してページネーションする
• データが上限を超える可能性がある場合は、cursors／nextCursor を使用して instances.query 呼び出しをページネーションする
• 書き込み操作をチャンクに分割し、instances.upsert 呼び出し 1 回あたり 1000 件にする
• 削除には、別途より厳しいタスクランナーを使用する
• デッドロックを防ぐため、cdfTaskRunner.schedule() のネストを避ける
• セマフォでラップした関数に対して Promise.all を使用し、生の API 呼び出しには使用しない
• テキストの照合には instances.search を、完全一致クエリには filter を使用する
• in フィルターの値を最大 1000 個のバッチに分割し、応答を統合する
• クエリ結果に対する LLM 呼び出しは上限（5／最大 50）を設けてキャッシュするか、実装しない
• 最新の上限については https://docs.cognite.com/cdf/dm/dm_reference/dm_limits_and_restrictions を参照する