Ingest Catalog index pipeline

The following sequence of steps ​illustrates the Catalog indexing pipeline implemented in Apache NiFi. The flow consists of mainly two stages:

1. Creating a Catalog document
2. Associating Catalog filters

This stage describes how to transform the Catalog data and load it into the Catalog index. It starts with running the following SQL to retrieve Catalog data from the Commerce database.

SELECT C.CATALOG_ID, C.IDENTIFIER, C.MEMBER_ID, D.NAME, D.SHORTDESCRIPTION, 
            COALESCE(D.LANGUAGE_ID, L.LANGUAGE_ID) LANGUAGE_ID, 
            L.LOCALENAME, T.MASTERCATALOG, F.STOREENT_ID
		  FROM LANGUAGE L, STORELANG F, STORECAT T, CATALOG C
          LEFT OUTER JOIN CATALOGDSC D ON(C.CATALOG_ID = D.CATALOG_ID AND D.LANGUAGE_ID = ${param.langId})
		 WHERE T.STOREENT_ID IN (SELECT RELATEDSTORE_ID FROM STOREREL WHERE STATE = 1 AND STRELTYP_ID = -4 AND STORE_ID = ${param.storeId})
            AND T.CATALOG_ID = C.CATALOG_ID  AND C.CATALOG_ID = ${param.catalogId}
		    AND F.STOREENT_ID = ${param.storeId} AND F.LANGUAGE_ID = L.LANGUAGE_ID AND L.LANGUAGE_ID = ${param.langId}	

For example code, see Stage 1 samples.

SELECT F.STOREENT_ID, F.CATALOG_ID, F.USAGE, F.CATFILTER_ID, F.IDENTIFIER, X.EXPRESSION_ID, X.MEMBER_ID,
	       CAST(X.QUERY AS VARCHAR) QUERY, X.TRADING_ID
		  FROM EXPRESSION X, CATFILTER F
	WHERE X.CATFILTER_ID = F.CATFILTER_ID
	AND F.CATALOG_ID IN (SELECT C.CATALOG_ID FROM STORECAT C​ WHERE C.MASTERCATALOG = 1
		                           AND C.STOREENT_ID IN (SELECT RELATEDSTORE_ID FROM STOREREL
			                                              WHERE STATE = 1 AND STRELTYP_ID = -4 AND STORE_ID = ${param.storeId}))
        AND F.STOREENT_ID IN ( SELECT RELATEDSTORE_ID FROM STOREREL WHERE STATE = 1 AND STRELTYP_ID = -32 AND STORE_ID = ${param.storeId} )
​    

Next, the result set is passed to the FindFiltersFromDatabase​ processor for transformation, using the following table to ​map the database field returned from the SQL above to an index field in the Catalog index:​For example code, see Stage 2 samples.

The following code is an example of the input data for the CreateCatalogDocumentFromDatabase processor:

{
  "CATALOG_ID": 10001,
  "IDENTIFIER": "Extended Sites Catalog Asset Store",
  "MEMBER_ID": 7000000000000001000,
  "NAME": "Extended Sites Catalog Asset Store",
  "SHORTDESCRIPTION": null,
  "LANGUAGE_ID": -1,
  "LOCALENAME": "en_US           ",
  "MASTERCATALOG": "1",
  "STOREENT_ID": 1
}

The CreateCatalogDocumentFromDatabase processor transforms the input data into the following output data:

{ "update": { "_id": "1--1-10001", "_index": ".auth.1.catalog.202006160325", 
"retry_on_conflict": 5, "_source": false } }
{
  "doc": {
    "identifier": {
      "catalog": "Extended Sites Catalog Asset Store",
      "specification": "catalog",
      "language": "en_US"
    },
    "name": {
      "normalized": "Extended Sites Catalog Asset Store",
      "raw": "Extended Sites Catalog Asset Store"
    },
    "id": {
      "catalog": "10001",
      "member": "7000000000000001000",
      "language": "-1",
      "store": "1"
    },
    "type": "master",
    "__meta": {
      "created": "2020-07-28T18:42:27.926Z",
      "modified": "2020-07-28T18:42:27.942Z",
      "version": {
        "min": 0,
        "max": 0
      }
    }
  },
  "doc_as_upsert": true
}

The following code is an example of the input data for the FindFiltersFromDatabase processor:

{
  "STOREENT_ID": 1,
  "CATALOG_ID": 10001,
  "USAGE": null,
  "CATFILTER_ID": 3074457345616679000,
  "IDENTIFIER": "TestCatalogFilter",
  "EXPRESSION_ID": 3074457345618259500,
  "MEMBER_ID": null,
  "QUERY": "( +*:* -parentCatgroup_id_search:\"10001_10006\")",
  "TRADING_ID": null
}

{ "update": { "_id": "1--1-10001", "_index": ".auth.1.catalog.202006160325", 
"retry_on_conflict": 5, "_source": false } }
{
  "doc": {
    "filters": [
      {
        "identifier": "TestCatalogFilter",
        "usage": "",
        "query": "( +*:* -path.tree:\"10006\")",
        "id": {
          "filter": "3074457345616679000",
          "expression": "3074457345618259500",
          "contract": "",
          "member": ""
        }
      }
    ],
    "__meta": {
      "modified": "2020-07-28T18:54:46.440Z"
    }
  }
}