集計パイプライン

MongoDBのためのOracle Database APIでは、MongoDB集計パイプライン(aggregateコマンド)がサポートされています。ほとんどのMongoDB集計ステージ、演算子および式がサポートされていますが、いくつかの制限があり、「機能サポート」に記載されています。

各集計パイプラインは、単一のSQL文に変換され、Oracle Databaseで実行されます。MongoDBとは異なり、Oracle Databaseでは、ソート、結合またはグループ化できるデータの量が制限されません。集計パイプラインは、複数のコレクションにまたがる数十億のドキュメントにわたるレポートまたは分析ワークロードに使用できます。パイプラインは、パラレル問合せ評価などのOracle Database機能の利点を活用して、処理時間を短縮することもできます。

例:

db.employees.insertMany([
    {"name" : "SMITH", "job" : "CLERK",    "sal" : 800},
    {"name" : "ALLEN", "job" : "SALESMAN", "sal" : 1600},
    {"name" : "WARD",  "job" : "SALESMAN", "sal" : 1250}
]);

db.employees.aggregate([
   {"$group": {"_id": "$job", "average_salary" : {"$avg":"$sal"}}},
   {"$sort": {"_id":1}}
 ]);
[
  { _id: 'CLERK', average_salary: 800 },
  { _id: 'SALESMAN', average_salary: 1425 }
]

この例では、employeesコレクションを作成し、3つのドキュメントを挿入し、job属性でドキュメントをグループ化します。個別ジョブごとに、平均給与が返され、ジョブ別に結果がソートされます。$groupステージおよび$sortステージを使用する集計パイプラインは、単一のSQL文として実行されます。生成されたSQLおよび問合せ計画を表示するには、explainコマンドを発行します。

db.employees.aggregate([ 
   { "$group": { "_id": "$job", "average_salary": { "$avg": "$sal" } } }, 
   { "$sort": { "_id": 1 } }
] ).explain();
{
  queryPlanner: {
    generatedSql: 'with \n' +
      '"Q1" ("ID", "DATA") as (select "RESID", "DATA" from "employees"), \n' +
      '"Q2" ("KEY0", "ACC0") as (\n' +
      '  select\n' +
      `    json_query("DATA", '$."job"' error on error json null on empty)\n` +
      '     as "KEY0", \n' +
      `    json_query("DATA", '$."sal"' error on error null on empty) as ACC0\n` +
      '  from "Q1" q\n' +
      '),\n' +
      ...
      'select "DATA" from "Q6"',
    winningPlan: ' Plan Hash Value  : 1694181348 \n' +
      '\n' +
      '----------------------------------------------------------------------------\n' +
      '| Id | Operation              | Name      | Rows | Bytes | Cost | Time     |\n' +
      '----------------------------------------------------------------------------\n' +
      '|  0 | SELECT STATEMENT       |           |    3 | 48954 |    5 | 00:00:01 |\n' +
      '|  1 |   SORT ORDER BY        |           |    3 | 48954 |    5 | 00:00:01 |\n' +
      '|  2 |    VIEW                |           |    3 | 48954 |    4 | 00:00:01 |\n' +
      '|  3 |     SORT GROUP BY      |           |    3 | 61461 |    4 | 00:00:01 |\n' +
      '|  4 |      TABLE ACCESS FULL | employees |    3 | 61461 |    3 | 00:00:01 |\n' +
      '----------------------------------------------------------------------------\n' 
  },
  ok: 1
}

この例では、explainコマンドを使用して、集計パイプラインに対して生成されたSQLおよび問合せ計画を表示します。集計パイプラインの内部生成SQLは、多くの場合、手書きのSQLよりも複雑で冗長です。ヒントを使用して、Oracle Databaseによる集計パイプラインの評価方法に影響を与えることができます。この例を続けて、$nativeヒントを適用してパラレル問合せの実行をリクエストします。

db.employees.aggregate(
  [ 
    { "$group": { "_id": "$job", "average_salary": { "$avg": "$sal" } } }, 
    { "$sort": { "_id": 1 } }
  ], 
  {"hint" : {$native:"PARALLEL"}} 
).explain();
{
  queryPlanner: {
    generatedSql: 'with \n' +
      '"Q1" ("ID", "DATA") as (select  /*+ PARALLEL */ "RESID", "DATA" from "employees"), \n' +
      ...
      'select  /*+ PARALLEL */ "DATA" from "Q6"',
    winningPlan: ' Plan Hash Value  : 3014829552 \n' +
      '\n' +
      '-------------------------------------------------------------------------------\n' +
      '| Id | Operation                 | Name      | Rows | Bytes | Cost | Time     |\n' +
      '-------------------------------------------------------------------------------\n' +
      '|  0 | SELECT STATEMENT          |           |    3 | 48954 |    4 | 00:00:01 |\n' +
      '|  1 |   SORT ORDER BY           |           |    3 | 48954 |    4 | 00:00:01 |\n' +
      '|  2 |    VIEW                   |           |    3 | 48954 |    3 | 00:00:01 |\n' +
      '|  3 |     SORT GROUP BY         |           |    3 | 61461 |    3 | 00:00:01 |\n' +
      '|  4 |      PX COORDINATOR       |           |      |       |      |          |\n' +
      '|  5 |       PX SEND QC (RANDOM) | :TQ10000  |    3 | 61461 |    2 | 00:00:01 |\n' +
      '|  6 |        PX BLOCK ITERATOR  |           |    3 | 61461 |    2 | 00:00:01 |\n' +
      '|  7 |         TABLE ACCESS FULL | employees |    3 | 61461 |    2 | 00:00:01 |\n' +
      '------------------------------------------------------------------------------\n' 
      ...

生成されたSQLにPARALLELヒントが含まれるようになり、実行計画により、文がパラレルに実行されることが確認されます。Oracle Autonomous Databaseで、コンシューマ・グループをHIGHまたはMEDIUMに設定して、パラレル実行を有効にします。パフォーマンスを参照してください。

MongoDBで定義されたステージに加えて、MongoDBのためのOracle APIでは、MongoDBでサポートされていない2つの追加ステージ($sqlおよび$external)がサポートされています。

$sql集計パイプライン・ステージ

$sqlステージを使用してOracle SQLおよびPL/SQLコードを実行できます。

例:

db.employees.insertMany([
    {"ename" : "SMITH", "job" : "CLERK",    "sal" : 800},
    {"ename" : "ALLEN", "job" : "SALESMAN", "sal" : 1600},
    {"ename" : "WARD",  "job" : "SALESMAN", "sal" : 1250}
  ]);

  db.aggregate([ {$sql :
    `select e.data.job, avg(e.data.sal) average
     from emps e
     group by e.data.job`
  } ]);
  [ 
    { JOB: 'CLERK', AVERAGE: 800 }, 
    { JOB: 'SALESMAN', AVERAGE: 1425 } 
  ]

この例では、$sqlステージを使用してselect問合せを実行し、各ジョブの従業員給与の平均を計算します。他のタイプのコマンドとは異なり、$sqlステージを使用して、JSONコレクションにないデータを作成およびアクセスできます。

例:

// create a relational table named "departments"
db.aggregate([{ $sql:`create table departments(deptno number, name varchar2(4000))`}]);

// insert two rows
db.aggregate([{ $sql:`insert into departments values (10, 'ACCOUNTING'), (20, 'RESEARCH')`}]);

// select the rows directly
db.aggregate([{ $sql:`select * from departments`}]);
[
  { DEPTNO: 10, NAME: 'ACCOUNTING' },
  { DEPTNO: 20, NAME: 'RESEARCH' }
]

この例では、表を作成し、2つの行を挿入して、データを選択します。

$sql構文

$sqlステージでは、次の構文を使用します:

{
  $sql    : {
    statement : <SQL statement>,
    binds : <variables>,
    dialect : <dialect>,
    format : <format>
  }
}

次の省略形も使用できます:

{
  $sql : <SQL statement>
}

省略形は、{$sql : {statement :<SQL statement>}}と同等です。

フィールド	説明
`statement`	実行するSQL文を指定します。(必須)
`binds`	SQL文内のプレースホルダ式(バインド変数)の値を指定します。(オプション) 次の「$sqlバインド」の項を参照してください。
`dialect`	SQL言語を指定します。サポートされている唯一の値は"oracle"です。他の値を指定すると、エラーが発生します。(オプション)
`format`	結果の形式を指定します。形式は"oracle"である必要があります。(オプション)
`resetSession`	MongoDBのためのOracle Database APIは、データベース接続をプールして再利用します。SQL文は、プールされた接続のセッション・ステートを変更できます(たとえば、"alter session"文を発行します)。デフォルトでは、接続を再利用した場合、セッション・ステートの変更は後続のコマンドに引き継がれます。セッションを後のコマンドで再利用できないようにするには、`resetSession`をtrueに設定します。コマンドがトランザクションの一部である場合、トランザクションが終了した後、セッションがリセットされます。(オプション)
`input`	`statement`が`FROM`句で参照して、前のステージのドキュメントにアクセスできる名前を指定します。このパラメータは、パイプライン内の`$sql`の前に別の集計ステージがあるか、開始コレクションが指定されている場合にのみ適用されます。(オプション)、デフォルト値"INPUT"

$sqlがパイプライン内の唯一のステージであり、開始コレクションがない場合、実行できるSQLのタイプに制限はありません。つまり、DDL、DMLおよびPL/SQLがサポートされています。

次に例を示します:

db.aggregate([
  {
    $sql: {
      statement: "update employees set salary = salary * 0.1"
    }
  }
]);

ただし、パイプラインがコレクションに対して実行される場合、または複数のステージが含まれている場合は、次の制限が適用されます:

SQLはSELECT文である必要があります。
SELECT文は、単一のJSON列を投影する必要があります。

$sqlステージでは、inputフィールドを使用して着信ドキュメントを参照することで、前のステージまたはコレクションからの出力を参照できます。前のステージのドキュメントでは、列名DATAを使用します。次に例を示します:

db.employees.aggregate([
  {$match: {job : "SALESMAN"}},
  {
    $sql: {
      statement: `select json_mergepatch(t.data, json{'updated':systimestamp}) from "MATCHED" t`,
      input: "MATCHED"
    }
  },
  {$out : "sales"}
])

この例では、$sqlを$matchおよび$outステージと組み合せます。SQLファンクションjson_mergepatchは、各ドキュメントにシステム・タイムスタンプを追加します。$sqlステージでは、"MATCHED"行ソースから選択して、前のステージのドキュメントを使用します。

$sql結果の形式

他のステージと同様に、$sqlはゼロ個以上のJSONオブジェクトを返します。オブジェクトの構造は、SQL文がSELECT文かどうかによって異なります。

SQL文がselect文の場合、問合せ結果の各行は、ステージ結果のJSONオブジェクトにマップされます。マッピングには次の2つのケースがあります:

ケース1: 問合せは、JSONデータを含む単一の列を返します。列は、JSON型、またはIS JSON制約のあるBLOB、CLOBまたはVARCHAR2列である必要があります。JSONドキュメントは、$sqlステージから直接返されます。次に例を示します:
```
scott> db.aggregate([{$sql:"select json{*} data from dept"}]);
  [
    { DEPTNO: 10, DNAME: 'ACCOUNTING', LOC: 'NEW YORK' },
    { DEPTNO: 20, DNAME: 'RESEARCH', LOC: 'DALLAS' },
    { DEPTNO: 30, DNAME: 'SALES', LOC: 'CHICAGO' },
    { DEPTNO: 40, DNAME: 'OPERATIONS', LOC: 'BOSTON' }
  ]
```

ケース2: 問合せで複数の列が生成される場合、列値を含むJSONオブジェクトが作成されます。列の別名は、対応する列値を持つオブジェクト・キー名として使用されます。次に例を示します:

scott> db.aggregate([{$sql:"select deptno, dname, loc from dept"}]);
  [
    { DEPTNO: 10, DNAME: 'ACCOUNTING', LOC: 'NEW YORK' },
    { DEPTNO: 20, DNAME: 'RESEARCH', LOC: 'DALLAS' },
    { DEPTNO: 30, DNAME: 'SALES', LOC: 'CHICAGO' },
    { DEPTNO: 40, DNAME: 'OPERATIONS', LOC: 'BOSTON' }
  ]

2の場合、SQL列値は新しいBSONドキュメントにマップされます。ケース2のBSON/SQL型マッピングは次のとおりです:

SQL型	BSON型
BLOB	IS JSON制約がある場合、値はJSON値として直接使用されます。それ以外の場合は、BSON RAWにマップされます。
CLOB	IS JSON制約がある場合、値はJSON値として直接使用されます。それ以外の場合は、BSON文字列にマップされます。
VARCHAR2	IS JSON制約がある場合、値はJSON値として直接使用されます。それ以外の場合は、BSON文字列にマップされます。
JSON	直接
TIMESTAMP	日付(UTCと仮定)
DATE	日付(UTCと仮定)
TIMESTAMPTZ	日付
NUMBER	スケールが0の場合、精度に応じてINT32またはINT64。それ以外の場合は、doubleになります。
RAW	バイナリ
その他の型	ERROR

SQL文がselect文でない場合、結果はresultという名前の単一のエントリを持つJSONオブジェクトになります。resultの値は、文の影響を受けた行数を示します。次に例を示します:

scott> db.aggregate([{$sql:`
      create table employee (name varchar2(4000), job varchar2(4000))
    `}]);
[ { result: 0 } ]

この例では、行が変更されていないため、create table文はresult値0を返します。

例:

db.aggregate([{$sql:
  {
    statement: "insert into employee values (:name, :job)",
    binds: {"name": "Bob", "job": "Programmer"}
  }
}]);
[ { result: 1 } ]

この例では、1行が挿入されたため、insert文はresult値1を返します。バッチ文の場合、resultは、各実行の行数を含む配列です。次に例を示します:

scott> db.aggregate([{$sql:
   {
    statement: "insert into employee values (:name, :job)",
    binds: [
      {"name": "John", "job": "Programmer"},
      {"name": "Jane", "job": "Manager"}
    ]
  }
 }]);
[ { result: [ 1, 1 ] } ]

この例では、resultに2つの値(1および1)が含まれています。これは、各実行が1つの行を挿入したためです。一部の文は複数の行に影響します。次に例を示します:

db.aggregate([{$sql:`
    delete from employee e 
    where e.job = 'Programmer'
`}]);
[ { result: 2 } ]

この例では、表から2つの行を削除し、戻り値のresultが2に設定されています。

$sqlバインド

bindsパラメータを使用して、SQL文のバインド変数(プレースホルダ式)の値を指定します。bindsの解釈は、値の形式によって異なります。

バインド・ケース1: 値は、次の属性を持つオブジェクトの配列です

フィールド	説明
`index`	位置指定バインド値の索引。この値は、nameパラメータと相互に排他的です。すべてのバインド値は、ユーザー名または索引のいずれかであり、両方を混在させることはできません。名前付きバインドは、$sqlがパイプライン内の唯一のステージである場合にのみサポートされます。
`name`	名前付きバインド値の名前。
`value`	バインド値
`dataType`	SQLバインド・タイプを指定します。許可される値とデフォルトについては、後述の「サポートされているバインド・タイプ」を参照してください。

次に例を示します:

db.aggregate([{
  $sql:{
     statement : `
       insert into employee(empno, ename) 
       values(:1,:2)
     `,
     binds : [
       {index:1, value:"E123"},
       {index:2, value:"JOHN DOE"}
     ]
  }
}]);

この例では、値"E123"が変数:1にバインドされ、値"JOHN DOE"が変数:2にバインドされています。

バインド・ケース2: 値はプリミティブ値の配列です

この場合、配列の各値は位置に基づいてバインドされます。配列の最初の値は最初のバインド変数にバインドされ、2番目の値は2番目のバインド変数にバインドされます。次に例を示します:

db.aggregate([{
  $sql:{
     statement : `
       insert into employee(empno, ename) 
       values(:1,:2)
     `,
     binds : [ "E123", "JOHN DOE" ]
  }
}]);

この例では、値"E123"が:1にバインドされ、値"JOHN DOE"が:2にバインドされています。この形式は、dataType属性を指定できない点を除き、ケース1の例と同等です。デフォルトのデータ型(この場合はVARCHAR2)が使用されます。

バインド・ケース3: 値はオブジェクトです

bindsが配列ではなくオブジェクトである場合、オブジェクト・キーおよび値は名前付きバインド変数の値を指定します。次に例を示します:

db.aggregate([{
  $sql:{
     statement : `
       insert into emp(empno, ename) 
       values(:empno,:ename)
     `,
     binds : {"empno":"E123", "ename":"JOHN DOE"}
  }
}]);

この例では、値"E123"が変数:empnoにバインドされており、値"JOHN DOE"が変数:enameにバインドされています。

バインド・ケース4: 値は、要素がケース1、ケース2またはケース3に準拠する配列です

この場合、Oracle Databaseは配列の各要素に対してSQL文を1回実行します。各配列要素は、その実行のバインド値を定義します。このアプローチでは、バッチの実行を効率的にサポートしています。バインド配列の使用は、$sqlを複数回実行するよりも効率的です。次に例を示します:

db.aggregate([{
  $sql:{
     statement : `
       insert into emp(empno, ename)
       values(:1,:2)
     `,
     binds : [
       ["E123", "JOHN DOE"],
       ["E456", "JANE DOE"]
     ]
  }
}]);

サポートされているバインド・タイプ

次の表に、特定のBSON値に対してサポートされているdataType属性の値を示します。

BSON値	サポートされているバインド・タイプ	デフォルトのバインド・タイプ
文字列	JSON、VARCHAR2	VARCHAR2
double	JSON、BINARY_DOUBLE	BINARY_DOUBLE
int32	JSON、NUMBER	NUMBER
int64	JSON、NUMBER	NUMBER
decimal128	JSON、NUMBER	NUMBER
true/false	JSON、VARCHAR2、BOOLEAN	BOOLEAN
objectid	JSON、RAW	RAW
binary	JSON、RAW	RAW
datetime	JSON、TIMESTAMP WITH TIME ZONE	TIMESTAMP WITH TIME ZONE
object	JSON、VARCHAR2	JSON
array	JSON、VARCHAR2	JSON
null	JSON、VARCHAR2、BINARY_DOUBLE、NUMBER、RAW、TIMESTAMP WITH TIMEZONE	VARCHAR2

BSON nullは、JSONを除くすべてのバインド・タイプのSQL nullにマップされます。バインド・タイプがJSONの場合、nullはJSON nullにマップされます。JSON列をSQL NULLに設定するには、VARCHAR2バインド・タイプを使用します。

dataTypeフィールドを使用する例:

result = db.aggregate([{
  $sql: {
    statement: `select json { 'example' : :foo} from dual`,
    binds: [{"name": "foo", value: {"hello":"world"}, dataType: "VARCHAR2"}]
  }
}]).toArray();

print(EJSON.stringify(result));
[{"example":"{\"hello\":\"world\"}"}]

この例では、オブジェクト{"hello":"world"}が:fooにVARCHAR2としてバインドされているため、結果に文字列値として表示されます。値をJSONとしてバインドすると、ネストされたオブジェクトとして返されます。

result = db.aggregate([{
  $sql: {
    statement: `select json { 'example' : :foo} from dual`,
    binds: [{"name": "foo", value: {"hello":"world"}, dataType: "JSON"}]
  }
}]).toArray();

print(EJSON.stringify(result));
[{"example":{"hello":"world"}}]

$external集計パイプライン・ステージ

$externalステージを使用して、データベース外のファイルに格納されているJSONドキュメントにアクセスします。

例:

db.aggregate([
  {$external:"https://raw.githubusercontent.com/oracle-samples/db-sample-schemas/main/order_entry/PurchaseOrders.dmp"}, 
  {$limit:1}, 
  {$project:{LineItems:0}}
])
[
  {
    PONumber: 1,
    Reference: 'MSULLIVA-20141102',
    Requestor: 'Martha Sullivan',
    User: 'MSULLIVA',
    CostCenter: 'A50',
    ShippingInstructions: {
      name: 'Martha Sullivan',
      Address: {
        street: '200 Sporting Green',
        city: 'South San Francisco',
        state: 'CA',
        zipCode: 99236,
        country: 'United States of America'
      },
      Phone: [ { type: 'Office', number: '979-555-6598' } ]
    },
    'Special Instructions': 'Surface Mail'
  }
]

この例では、Oracle GitHubサンプル・リポジトリに格納されている行区切りテキスト・ファイル(PurchaseOrders.dmp)からJSONオブジェクトを読み取ります。$externalと、ファイル内の最初のオブジェクトのみを返す$limit、およびオブジェクトからLineItems属性を削除する$projectを組み合せます。

$external構文

{
  $external : {
    location : <URL or file name>,
    directory : <database directory name>,
    credential : <credential name>
    path : <SQL/JSON path expression>
  }
}

次の省略構文を使用することもできます:

{
  $external : <pre-authenticated URI>
}

この省略形は、{$external : {location: <pre-authenticated URI>}}と同等であり、パブリック・アクセスを持つリソースにのみ適用されます。

$externalステージでは、ORACLE_BIGDATAアクセス・ドライバに基づいてインライン外部表定義を使用するSQL文が生成されます。$externalフィールドは、ORACLE_BIGDATAドライバのJSONパラメータにマップされます。

フィールド	説明
`location`	アクセスするJSONファイルの場所を指定します。(必須) 参照: Location句
`directory`	ファイルが格納されているディレクトリを指定します。(オプション) 参照: Location句
`credential`	オブジェクト・ストア内のデータ・ファイルへのアクセスに使用される資格証明オブジェクトを識別します。(オプション) アクセス・パラメータcom.oracle.bigdata.credential.nameを参照してください
`path`	ステージが返すネストされたオブジェクトへのパスを指定します。(オプション) com.oracle.bigdata.json.pathアクセス・パラメータを参照してください

次の例は、$externalを使用して、Autonomous Databaseでオブジェクト・ストレージ内の保護されたファイルにアクセスする方法を示しています。

db.aggregate([{ $sql: 
     `BEGIN
     DBMS_CLOUD.CREATE_CREDENTIAL(
         credential_name => 'MYCRED',
         username        => 'myuser@oracle.com',
         password        => 'XXXXXXXXX'
     );
     END;`
}]);

db.aggregate([
  {
    $external: {
      location:"https://objectstorage.ca-toronto-1.oraclecloud.com/n/yz6dzkrqow85/b/private-demo/o/movies.json", 
      credential:"MYCRED"
    }
  }, 
  {$limit:1}
])
[
  {
    studio: null,
    title: "'Gator Bait II: Cajun Justice",
    summary: "'  Gator Bait II: Cajun Justice is a 1988 sequel to the 1974 film 'Gator Bait , written, produced and directed by Beverly Sebastian and Ferd Sebastian. Largely ignored upon release, the film received a second life on cable television and home video.",
    sku: 'COO3790',
    list_price: 3.99,
    year: 1988,
  }
]

この例では、最初にDBMS_CLOUD.CREATE_CREDENTIALを使用してオブジェクト・ストレージの資格証明を作成し、その資格証明を使用してオブジェクト・ストレージのJSONファイルにアクセスします。詳細は、オブジェクト・ストアの資格証明の作成方法を参照してください。

次の例は、$externalがディレクトリ内のJSONファイルにアクセスする方法を示しています。

db.aggregate([{$sql:"CREATE DIRECTORY DEMODIR AS '/tmp/demo'"}]);
[ { result: 0 } ]

db.aggregate([
     {
         $external: {
             location:"movie.json", 
             directory:"DEMODIR"
         }
     }, 
     {
         $limit:1
     }
]);
[
  {
    studio: null,
    title: "'Gator Bait II: Cajun Justice",
    summary: "'  Gator Bait II: Cajun Justice is a 1988 sequel to the 1974 film 'Gator Bait , written, produced and directed by Beverly Sebastian and Ferd Sebastian. Largely ignored upon release, the film received a second life on cable television and home video.",
    sku: 'COO3790',
    list_price: 3.99,
    year: 1988
  }
]

この例では、/tmp/demoを指すDEMODIRという名前のディレクトリを作成し、そのディレクトリ内のファイルmovie.jsonにアクセスします。