Merge pull request #3683 from IBM/robin-citus-1

Citus distribution support and async/remote indexing

Author: lmsurpre

build/docker/deploySchemaAndTenant.sh

@@ -21,7 +21,7 @@ while [ "$not_ready" == "true" ]
 do
   EXIT_CODE="-1"
   java -jar schema/fhir-persistence-schema-*-cli.jar \
-    --prop-file db2.properties --schema-name FHIRDATA --create-schemas | tee -a ${TMP_FILE}
+    --db-type db2 --prop-file db2.properties --schema-name FHIRDATA --create-schemas | tee -a ${TMP_FILE}
   EXIT_CODE="${PIPESTATUS[0]}"
   LOG_OUT=`cat ${TMP_FILE}`
   if [[ "$EXIT_CODE" == "0" ]]
@@ -54,18 +54,18 @@ then
 fi
 
 java -jar schema/fhir-persistence-schema-*-cli.jar \
-  --prop-file db2.properties --schema-name FHIRDATA --update-schema --pool-size 2
+  --db-type db2 --prop-file db2.properties --schema-name FHIRDATA --update-schema --pool-size 2
 
 java -jar schema/fhir-persistence-schema-*-cli.jar \
-  --prop-file db2.properties --schema-name FHIRDATA --grant-to FHIRSERVER --pool-size 2
+  --db-type db2 --prop-file db2.properties --schema-name FHIRDATA --grant-to FHIRSERVER --pool-size 2
 
 java -jar schema/fhir-persistence-schema-*-cli.jar \
-  --prop-file db2.properties --schema-name FHIRDATA --allocate-tenant default --pool-size 2
+  --db-type db2 --prop-file db2.properties --schema-name FHIRDATA --allocate-tenant default --pool-size 2
 
 # The regex in the following command will output the capture group between "key=" and "]"
 # With GNU grep, the following would work as well:  grep -oP 'key=\K\S+(?=])'
 tenantKey=$(java -jar schema/fhir-persistence-schema-*-cli.jar \
-  --prop-file db2.properties --schema-name FHIRDATA --add-tenant-key default 2>&1 \
+  --db-type db2 --prop-file db2.properties --schema-name FHIRDATA --add-tenant-key default 2>&1 \
   | grep "key=" | sed -e 's/.*key\=\(.*\)\].*/\1/')
 
 # Creating a backup file is the easiest way to make in-place sed portable across OSX and Linux

build/docker/updateSchema.sh

@@ -13,17 +13,17 @@ cd ${DIR}
 # For #1366 the migration hits deadlock issues if run in parallel, so
 # to avoid this, serialize the steps using --pool-size 1
 java -jar schema/fhir-persistence-schema-*-cli.jar \
-  --prop-file db2.properties --schema-name FHIRDATA --update-schema \
+  --db-type db2 --prop-file db2.properties --schema-name FHIRDATA --update-schema \
   --pool-size 1
 
 # Rerun grants to cover any new tables added by the above migration step 
 java -jar schema/fhir-persistence-schema-*-cli.jar \
-  --prop-file db2.properties --schema-name FHIRDATA --grant-to FHIRSERVER --pool-size 2
+  --db-type db2 --prop-file db2.properties --schema-name FHIRDATA --grant-to FHIRSERVER --pool-size 2
 
 # And make sure that the new tables have partitions for existing tenants
 java -jar schema/fhir-persistence-schema-*-cli.jar \
-  --prop-file db2.properties --refresh-tenants
+  --db-type db2 --prop-file db2.properties --refresh-tenants
 
 java -jar schema/fhir-persistence-schema-*-cli.jar \
-  --prop-file db2.properties --schema-name FHIRDATA --grant-to FHIRSERVER \
+  --db-type db2 --prop-file db2.properties --schema-name FHIRDATA --grant-to FHIRSERVER \
   --pool-size 20 

build/migration/bin/6_current-reindex.sh

@@ -23,7 +23,7 @@ run_reindex(){
         DATE_ISO=$(date +%Y-%m-%dT%H:%M:%SZ)
         status=$(curl -k -X POST -o reindex.json -i -w '%{http_code}' -u 'fhiruser:change-password' 'https://localhost:9443/fhir-server/api/v4/$reindex' \
             -H 'Content-Type: application/fhir+json' -H 'X-FHIR-TENANT-ID: default' \
-            -d "{\"resourceType\": \"Parameters\",\"parameter\":[{\"name\":\"resourceCount\",\"valueInteger\":100},{\"name\":\"tstamp\",\"valueString\":\"${DATE_ISO}\"}]}")
+            -d "{\"resourceType\": \"Parameters\",\"parameter\":[{\"name\":\"resourceCount\",\"valueInteger\":100},{\"name\":\"tstamp\",\"valueString\":\"${DATE_ISO}\"},{\"name\":\"force\",\"valueBoolean\":true}]}")
         echo "Status: ${status}"
 
         while [ $status -ne 200 ]
@@ -57,7 +57,7 @@ run_reindex(){
             fi
             status=$(curl -k -X POST -o reindex.json -i -w '%{http_code}' -u 'fhiruser:change-password' 'https://localhost:9443/fhir-server/api/v4/$reindex' \
                 -H 'Content-Type: application/fhir+json' -H 'X-FHIR-TENANT-ID: default' \
-                -d "{\"resourceType\": \"Parameters\",\"parameter\":[{\"name\":\"resourceCount\",\"valueInteger\":100},{\"name\":\"tstamp\",\"valueString\":\"${DATE_ISO}\"}]}")
+                -d "{\"resourceType\": \"Parameters\",\"parameter\":[{\"name\":\"resourceCount\",\"valueInteger\":100},{\"name\":\"tstamp\",\"valueString\":\"${DATE_ISO}\"},{\"name\":\"force\",\"valueBoolean\":true}]}")
             echo "Status: ${status}"
         done
     fi
@@ -77,4 +77,4 @@ run_reindex "${1}"
 popd > /dev/null
 
 # EOF
-###############################################################################
+###############################################################################

docs/src/pages/guides/FHIRSearchConfiguration.md

@@ -217,6 +217,7 @@ By default, the operation will select 10 resources and re-extract their search p
 |----|----|-----------|
 |`tstamp`|string|Reindex only resources not previously reindexed since this timestamp. Format as a date YYYY-MM-DD or time YYYY-MM-DDTHH:MM:SSZ.|
 |`resourceCount`|integer|The maximum number of resources to reindex in this call. If this number is too large, the processing time might exceed the transaction timeout and fail.|
+|`force`|boolean|Force the parameters to be replaced even if the parameter hash matches. This is only required following a schema migration which changes how the parameters are stored in the database.|
 
 The IBM FHIR Server tracks when a resource was last reindexed and only resources with a reindex_tstamp value less than the given tstamp parameter will be processed. When a resource is reindexed, its reindex_tstamp is set to the given tstamp value. In most cases, using the current date (for example "2020-10-27") is the best option for this value.
 

docs/src/pages/guides/FHIRServerUsersGuide.md

@@ -27,6 +27,7 @@ permalink: /FHIRServerUsersGuide/
   * [4.10 Bulk data operations](#410-bulk-data-operations)
   * [4.11 Audit logging service](#411-audit-logging-service)
   * [4.12 FHIR REST API](#412-fhir-rest-api)
+  * [4.13 Remote Index Service](#413-remote-index-service)
 - [5 Appendix](#5-appendix)
   * [5.1 Configuration properties reference](#51-configuration-properties-reference)
   * [5.2 Keystores, truststores, and the FHIR server](#52-keystores-truststores-and-the-fhir-server)
@@ -2070,6 +2071,10 @@ For example, consider a FHIR Server that is listening at https://fhir:9443/fhir-
 
 The originalRequestUriHeader is expected to contain the full path of the original request. Values with no scheme (e.g. `https://`) will be handled like relative URLs, but full URL values (including scheme, hostname, optional port, and path) are recommended. Query string values can be included in the header value but will be ignored by the server; the server will use the query string of the actual request to process the request.
 
+### 4.13 Remote Index Service
+
+To use the experimental remote index service feature, see the instructions documented in the [fhir-remote-index](https://github.com/IBM/FHIR/tree/main/operation/fhir-remote-index/README.md) project.
+
 # 5 Appendix
 
 ## 5.1 Configuration properties reference
@@ -2260,7 +2265,17 @@ This section contains reference information about each of the configuration prop
 |`fhirServer/operations/membermatch/strategy`|string|The key identifying the Member Match strategy|
 |`fhirServer/operations/membermatch/extendedProps`|object|The extended options for the extended member match implementation|
 |`fhirServer/operations/everything/includeTypes`|list|The list of related resource types to include alongside the patient compartment resource types. Instances of these resource types will only be returned when they are referenced from one or more resource instances from the target patient compartment. Example values are like `Location`, `Medication`, `Organization`, and `Practitioner`|
-
+|`fhirServer/remoteIndexService/type`|string| The type of service used to send remote index messages. Only `kafka` is currently supported|
+|`fhirServer/remoteIndexService/instanceIdentifier`|string| A UUID or other identifier unique to this cluster of IBM FHIR Servers |
+|`fhirServer/remoteIndexService/kafka/mode`|string| Current operation mode of the service. Specify `ACTIVE` to use the service|
+|`fhirServer/remoteIndexService/kafka/topicName`|string| The Kafka topic name. Typically `FHIR_REMOTE_INDEX` |
+|`fhirServer/remoteIndexService/kafka/connectionProperties/bootstrap.servers`|string| Bootstrap servers for the Kafka service |
+|`fhirServer/remoteIndexService/kafka/connectionProperties/sasl.jaas.config`|string| Kafka service authentication |
+|`fhirServer/remoteIndexService/kafka/connectionProperties/sasl.mechanism`|string| Kafka service authentication|
+|`fhirServer/remoteIndexService/kafka/connectionProperties/security.protocol`|string| Kafka service security |
+|`fhirServer/remoteIndexService/kafka/connectionProperties/ssl.protocol`|string| Kafka service SSL configuration |
+|`fhirServer/remoteIndexService/kafka/connectionProperties/ssl.enabled.protocols`|string| Kafka service SSL configuration |
+|`fhirServer/remoteIndexService/kafka/connectionProperties/ssl.endpoint.identification.algorithm`|string| Kafka service SSL configuration |
 
 ### 5.1.2 Default property values
 | Property Name                 | Default value   |
@@ -2406,6 +2421,17 @@ This section contains reference information about each of the configuration prop
 |`fhirServer/operations/membermatch/strategy`|default|
 |`fhirServer/operations/membermatch/extendedProps`|empty|
 |`fhirServer/operations/everything/includeTypes`|null|
+|`fhirServer/remoteIndexService/type`|null|
+|`fhirServer/remoteIndexService/instanceIdentifier`|null|
+|`fhirServer/remoteIndexService/kafka/mode`|null|
+|`fhirServer/remoteIndexService/kafka/topicName`|null|
+|`fhirServer/remoteIndexService/kafka/connectionProperties/bootstrap.servers`|null|
+|`fhirServer/remoteIndexService/kafka/connectionProperties/sasl.jaas.config`|null|
+|`fhirServer/remoteIndexService/kafka/connectionProperties/sasl.mechanism`|null|
+|`fhirServer/remoteIndexService/kafka/connectionProperties/security.protocol`|null|
+|`fhirServer/remoteIndexService/kafka/connectionProperties/ssl.protocol`|null|
+|`fhirServer/remoteIndexService/kafka/connectionProperties/ssl.enabled.protocols`|null|
+|`fhirServer/remoteIndexService/kafka/connectionProperties/ssl.endpoint.identification.algorithm`|null|
 
 ### 5.1.3 Property attributes
 Depending on the context of their use, config properties can be:
@@ -2588,6 +2614,17 @@ Cases where that behavior is not supported are marked below with an `N` in the `
 |`fhirServer/operations/membermatch/strategy`|Y|Y|Y|
 |`fhirServer/operations/membermatch/extendedProps`|Y|Y|Y|
 |`fhirServer/operations/everything/includeTypes`|Y|Y|N|
+|`fhirServer/remoteIndexService/type`|N|N|N|
+|`fhirServer/remoteIndexService/instanceIdentifier`|N|N|N|
+|`fhirServer/remoteIndexService/kafka/mode`|N|N|N|
+|`fhirServer/remoteIndexService/kafka/topicName`|N|N|N|
+|`fhirServer/remoteIndexService/kafka/connectionProperties/bootstrap.servers`|N|N|N|
+|`fhirServer/remoteIndexService/kafka/connectionProperties/sasl.jaas.config`|N|N|N|
+|`fhirServer/remoteIndexService/kafka/connectionProperties/sasl.mechanism`|N|N|N|
+|`fhirServer/remoteIndexService/kafka/connectionProperties/security.protocol`|N|N|N|
+|`fhirServer/remoteIndexService/kafka/connectionProperties/ssl.protocol`|N|N|N|
+|`fhirServer/remoteIndexService/kafka/connectionProperties/ssl.enabled.protocols`|N|N|N|
+|`fhirServer/remoteIndexService/kafka/connectionProperties/ssl.endpoint.identification.algorithm`|N|N|N|
 
 ## 5.2 Keystores, truststores, and the IBM FHIR server
 

fhir-bucket/src/main/java/com/ibm/fhir/bucket/app/Main.java

@@ -57,10 +57,13 @@
 import com.ibm.fhir.database.utils.api.IDatabaseAdapter;
 import com.ibm.fhir.database.utils.api.IDatabaseTranslator;
 import com.ibm.fhir.database.utils.api.ILeaseManagerConfig;
+import com.ibm.fhir.database.utils.api.ISchemaAdapter;
 import com.ibm.fhir.database.utils.api.ITransaction;
 import com.ibm.fhir.database.utils.api.ITransactionProvider;
+import com.ibm.fhir.database.utils.api.SchemaApplyContext;
 import com.ibm.fhir.database.utils.api.UniqueConstraintViolationException;
 import com.ibm.fhir.database.utils.common.JdbcConnectionProvider;
+import com.ibm.fhir.database.utils.common.PlainSchemaAdapter;
 import com.ibm.fhir.database.utils.db2.Db2Adapter;
 import com.ibm.fhir.database.utils.db2.Db2PropertyAdapter;
 import com.ibm.fhir.database.utils.db2.Db2Translator;
@@ -147,6 +150,9 @@ public class Main {
     // The adapter configured for the type of database we're using
     private IDatabaseAdapter adapter;
 
+    // The (plain) schema adapter which wraps the database adapter
+    private ISchemaAdapter schemaAdapter;
+
     // The number of threads to use for the schema creation step
     private int createSchemaThreads = 1;
 
@@ -639,6 +645,7 @@ public void configure() {
                 setupDerbyRepository();
                 break;
             case POSTGRESQL:
+            case CITUS:
                 setupPostgresRepository();
                 break;
             }
@@ -669,6 +676,7 @@ public void setupDerbyRepository() {
         this.connectionPool = new PoolConnectionProvider(cp, connectionPoolSize);
         this.connectionPool.setCloseOnAnyError();
         this.adapter = new DerbyAdapter(connectionPool);
+        this.schemaAdapter = new PlainSchemaAdapter(adapter);
         this.transactionProvider = new SimpleTransactionProvider(connectionPool);
     }
 
@@ -692,6 +700,7 @@ public void setupDb2Repository() {
         IConnectionProvider cp = new JdbcConnectionProvider(translator, propertyAdapter);
         this.connectionPool = new PoolConnectionProvider(cp, connectionPoolSize);
         this.adapter = new Db2Adapter(connectionPool);
+        this.schemaAdapter = new PlainSchemaAdapter(adapter);
         this.transactionProvider = new SimpleTransactionProvider(connectionPool);
     }
 
@@ -715,6 +724,7 @@ public void setupPostgresRepository() {
         IConnectionProvider cp = new JdbcConnectionProvider(translator, propertyAdapter);
         this.connectionPool = new PoolConnectionProvider(cp, connectionPoolSize);
         this.adapter = new PostgresAdapter(connectionPool);
+        this.schemaAdapter = new PlainSchemaAdapter(adapter);
         this.transactionProvider = new SimpleTransactionProvider(connectionPool);
     }
 
@@ -732,7 +742,7 @@ protected VersionHistoryService createVersionHistoryService() {
         // Create the version history table if it doesn't yet exist
         try (ITransaction tx = transactionProvider.getTransaction()) {
             try {
-                CreateVersionHistory.createTableIfNeeded(schemaName, this.adapter);
+                CreateVersionHistory.createTableIfNeeded(schemaName, this.schemaAdapter);
             } catch (Exception x) {
                 logger.log(Level.SEVERE, "failed to create version history table", x);
                 tx.setRollbackOnly();
@@ -761,8 +771,8 @@ public void bootstrapDb() {
             try (ITransaction tx = transactionProvider.getTransaction()) {
                 try {
                     adapter.createSchema(schemaName);
-                    CreateControl.createTableIfNeeded(schemaName, adapter);
-                    CreateWholeSchemaVersion.createTableIfNeeded(schemaName, adapter);
+                    CreateControl.createTableIfNeeded(schemaName, schemaAdapter);
+                    CreateWholeSchemaVersion.createTableIfNeeded(schemaName, schemaAdapter);
                     success = true;
                 } catch (Exception x) {
                     logger.log(Level.SEVERE, "failed to create schema management tables", x);
@@ -822,7 +832,8 @@ private void buildSchema() {
         TaskService taskService = new TaskService();
         ExecutorService pool = Executors.newFixedThreadPool(this.createSchemaThreads);
         ITaskCollector collector = taskService.makeTaskCollector(pool);
-        pdm.collect(collector, adapter, this.transactionProvider, vhs);
+        SchemaApplyContext context = SchemaApplyContext.getDefault();
+        pdm.collect(collector, schemaAdapter, context, this.transactionProvider, vhs);
 
         // FHIR in the hole!
         logger.info("Starting schema updates");
@@ -845,7 +856,7 @@ private void buildSchema() {
             try {
                 Set<String> resourceTypes = ResourceTypeHelper.getR4bResourceTypesFor(FHIRVersionParam.VERSION_43);
 
-                if (adapter.getTranslator().getType() == DbType.POSTGRESQL) {
+                if (adapter.getTranslator().isFamilyPostgreSQL()) {
                     // Postgres doesn't support batched merges, so we go with a simpler UPSERT
                     MergeResourceTypesPostgres mrt = new MergeResourceTypesPostgres(schemaName, resourceTypes);
                     adapter.runStatement(mrt);

fhir-bucket/src/main/java/com/ibm/fhir/bucket/persistence/AddBucketPath.java

@@ -17,7 +17,6 @@
 import com.ibm.fhir.database.utils.api.IDatabaseSupplier;
 import com.ibm.fhir.database.utils.api.IDatabaseTranslator;
 import com.ibm.fhir.database.utils.common.DataDefinitionUtil;
-import com.ibm.fhir.database.utils.model.DbType;
 
 /**
  * DAO to encapsulate all the SQL/DML used to retrieve and persist data
@@ -57,7 +56,7 @@ public Long run(IDatabaseTranslator translator, Connection c) {
         // try the old-fashioned way and handle duplicate key
         final String bucketPaths = DataDefinitionUtil.getQualifiedName(schemaName, "bucket_paths");
         final String dml;
-        if (translator.getType() == DbType.POSTGRESQL) {
+        if (translator.isFamilyPostgreSQL()) {
             // For POSTGRES, if a statement fails it causes the whole transaction
             // to fail, so we need turn this into an UPSERT
             dml = "INSERT INTO " + bucketPaths + "(bucket_name, bucket_path) VALUES (?,?) ON CONFLICT(bucket_name, bucket_path) DO NOTHING";

fhir-bucket/src/main/java/com/ibm/fhir/bucket/persistence/AddResourceBundle.java

@@ -20,7 +20,6 @@
 import com.ibm.fhir.database.utils.api.IDatabaseSupplier;
 import com.ibm.fhir.database.utils.api.IDatabaseTranslator;
 import com.ibm.fhir.database.utils.common.DataDefinitionUtil;
-import com.ibm.fhir.database.utils.model.DbType;
 
 /**
  * DAO to encapsulate all the SQL/DML used to retrieve and persist data
@@ -76,7 +75,7 @@ public ResourceBundleData run(IDatabaseTranslator translator, Connection c) {
         int version = 1;
         final String resourceBundles = DataDefinitionUtil.getQualifiedName(schemaName, "resource_bundles");
         final String dml;
-        if (translator.getType() == DbType.POSTGRESQL) {
+        if (translator.isFamilyPostgreSQL()) {
             // For PostgresSQL, make sure we don't break the current transaction
             // if the statement fails...annoying
             dml = "INSERT INTO " + resourceBundles + "("