fix: 🐛 Floats ending in .0 now being parsed as float type

Float numbers ending in .0 where being parsed as integers in an
OpenSearch instance when using the @searchable directive. If this data
was the first being uploaded to the instance, searching queries wouldn't
work as expected. The change creates a JSON file which is uploaded with
the Python streaming function to lambda. The JSON has all the schema
data types of the @searchable model fields. A modified version of the
Python streaming function enables the DynamoDB event fields to be
correctly parsed to the correct data type described in the schema.

✅ Closes: aws-amplify#866

packages/amplify-graphql-searchable-transformer/package.json

@@ -21,7 +21,7 @@
     "access": "public"
   },
   "scripts": {
-    "build": "tsc && cd streaming-lambda && bestzip --force node ../lib/streaming-lambda.zip python_streaming_function.py",
+    "build": "tsc && cd streaming-lambda && cp python_streaming_function.py ../lib",
     "watch": "tsc -w",
     "clean": "rimraf ./lib",
     "test": "jest"
@@ -37,12 +37,14 @@
     "@aws-cdk/aws-iam": "~1.124.0",
     "@aws-cdk/aws-lambda": "~1.124.0",
     "@aws-cdk/core": "~1.124.0",
+    "adm-zip": "0.5.9",
     "graphql": "^14.5.8",
     "graphql-mapping-template": "4.20.5",
     "graphql-transformer-common": "4.24.0"
   },
   "devDependencies": {
     "@aws-cdk/assert": "~1.124.0",
+    "@types/adm-zip": "^0.5.0",
     "@types/node": "^12.12.6"
   },
   "jest": {

packages/amplify-graphql-searchable-transformer/src/cdk/create-streaming-lambda.ts

@@ -9,7 +9,60 @@ import {
   Effect, IRole, Policy, PolicyStatement, Role, ServicePrincipal,
 } from '@aws-cdk/aws-iam';
 import { ResourceConstants, SearchableResourceIDs } from 'graphql-transformer-common';
+import {
+  ObjectTypeDefinitionNode,
+  TypeNode,
+} from 'graphql';
 import * as path from 'path';
+import * as fs from 'fs';
+import AdmZip from 'adm-zip';
+
+const DATA_TYPE_SCHEMA_FILENAME = 'schema_datatypes.json';
+const STREAMING_FUNCTION_FILENAME = 'python_streaming_function.py';
+const STREAMING_LAMBDA_ZIP_FILENAME = 'streaming-lambda.zip';
+interface AttributeTypes {
+  [attribute: string]: string;
+}
+interface SchemaDataTypes {
+  [modelName: string]: AttributeTypes;
+}
+
+const findNamedType = (typeNode: TypeNode) : string => {
+  switch (typeNode.kind) {
+    case 'NamedType':
+      return typeNode.name.value;
+    case 'ListType':
+    case 'NonNullType':
+      return findNamedType(typeNode.type);
+    default:
+      throw new Error(`Unknown type ${typeNode}`);
+  }
+};
+
+const generateSchemaDataTypes = (searchableObjectTypeDefinitions: { node: ObjectTypeDefinitionNode; fieldName: string; }[]): void => {
+  const schemaDataTypes: SchemaDataTypes = {};
+  for (const def of searchableObjectTypeDefinitions) {
+    const modelName = def.node.name.value.toLowerCase();
+
+    const attributeTypes: AttributeTypes = {};
+    def.node.fields?.forEach((f) => {
+      attributeTypes[f.name.value] = findNamedType(f.type);
+    });
+    schemaDataTypes[modelName] = attributeTypes;
+  }
+
+  // Paths to export JSON file and lambda function script
+  const libPath = path.join(__dirname, '..', '..', 'lib');
+  const schemaPath = path.join(libPath, DATA_TYPE_SCHEMA_FILENAME);
+  const streamingFunctionPath = path.join(libPath, STREAMING_FUNCTION_FILENAME);
+  fs.writeFileSync(schemaPath, JSON.stringify(schemaDataTypes));
+
+  // Zip the file
+  const zip = new AdmZip();
+  zip.addLocalFile(schemaPath);
+  zip.addLocalFile(streamingFunctionPath);
+  zip.writeZip(path.join(libPath, STREAMING_LAMBDA_ZIP_FILENAME));
+};
 
 export const createLambda = (
   stack: Stack,
@@ -19,6 +72,7 @@ export const createLambda = (
   endpoint: string,
   isProjectUsingDataStore: boolean,
   region: string,
+  searchableObjectTypeDefinitions: { node: ObjectTypeDefinitionNode; fieldName: string }[],
 ): IFunction => {
   const { OpenSearchStreamingLambdaFunctionLogicalID } = ResourceConstants.RESOURCES;
   const { OpenSearchStreamingLambdaHandlerName, OpenSearchDebugStreamingLambda } = ResourceConstants.PARAMETERS;
@@ -29,11 +83,13 @@ export const createLambda = (
     OPENSEARCH_USE_EXTERNAL_VERSIONING: isProjectUsingDataStore.toString(),
   };
 
+  generateSchemaDataTypes(searchableObjectTypeDefinitions);
+
   return apiGraphql.host.addLambdaFunction(
     OpenSearchStreamingLambdaFunctionLogicalID,
     `functions/${OpenSearchStreamingLambdaFunctionLogicalID}.zip`,
     parameterMap.get(OpenSearchStreamingLambdaHandlerName)!.valueAsString,
-    path.resolve(__dirname, '..', '..', 'lib', 'streaming-lambda.zip'),
+    path.resolve(__dirname, '..', '..', 'lib', STREAMING_LAMBDA_ZIP_FILENAME),
     Runtime.PYTHON_3_8,
     [
       LayerVersion.fromLayerVersionArn(

packages/amplify-graphql-searchable-transformer/src/graphql-searchable-transformer.ts

@@ -2,7 +2,8 @@ import {
   TransformerPluginBase,
   generateGetArgumentsInput,
   InvalidDirectiveError,
-  MappingTemplate, DirectiveWrapper,
+  MappingTemplate,
+  DirectiveWrapper,
 } from '@aws-amplify/graphql-transformer-core';
 import {
   DataSourceProvider,
@@ -71,8 +72,11 @@ const getTable = (context: TransformerContextProvider, definition: ObjectTypeDef
 
 const getNonKeywordFields = (def: ObjectTypeDefinitionNode): Expression[] => {
   const nonKeywordTypeSet = new Set(nonKeywordTypes);
-  return def.fields?.filter((field) => nonKeywordTypeSet.has(getBaseType(field.type))
-    && !DATASTORE_SYNC_FIELDS.includes(field.name.value)).map((field) => str(field.name.value)) || [];
+  return (
+    def.fields
+      ?.filter((field) => nonKeywordTypeSet.has(getBaseType(field.type)) && !DATASTORE_SYNC_FIELDS.includes(field.name.value))
+      .map((field) => str(field.name.value)) || []
+  );
 };
 
 /**
@@ -327,6 +331,7 @@ export class SearchableModelTransformer extends TransformerPluginBase {
       domain.domainEndpoint,
       isProjectUsingDataStore,
       region,
+      this.searchableObjectTypeDefinitions,
     );
 
     for (const def of this.searchableObjectTypeDefinitions) {
@@ -362,7 +367,7 @@ export class SearchableModelTransformer extends TransformerPluginBase {
         MappingTemplate.s3MappingTemplateFromString(
           requestTemplate(
             attributeName,
-            getNonKeywordFields((context.output.getObject(type))as ObjectTypeDefinitionNode),
+            getNonKeywordFields(context.output.getObject(type) as ObjectTypeDefinitionNode),
             context.isProjectUsingDataStore(),
             openSearchIndexName,
             keyFields,

packages/amplify-graphql-searchable-transformer/streaming-lambda/python_streaming_function.py

@@ -60,6 +60,26 @@ def __init__(self, status_code, payload):
         Exception.__init__(
             self, 'Searchable_Exception: status_code={}, payload={}'.format(status_code, payload))
 
+# Custom mapper to match Python types to GraphQL Schema types
+def map_to_gql_types(fields, table):
+    mapped_fields = {}
+    # Get GraphQL schema types
+    schema_types = getGQLSchema(table)
+
+    for field in fields:
+        v = fields[field]
+        data_type = schema_types[field] if field in schema_types else None
+        if data_type == 'Float' and isinstance(v, int):
+            mapped_fields[field] = float(v)
+        else:
+            mapped_fields[field] = v
+    return mapped_fields
+
+# Gets schema from json file
+def getGQLSchema(table):
+    with open('schema_datatypes.json') as f:
+        schema = json.load(f)
+    return schema[table]
 
 # Low-level POST data to Amazon OpenSearch Service generating a Sigv4 signed request
 def post_data_to_opensearch(payload, region, creds, host, path, method='POST', proto='https://'):
@@ -188,6 +208,11 @@ def _lambda_handler(event, context):
         logger.debug(image_name + ': %s', ddb[image_name])
         # Deserialize DynamoDB type to Python types
         doc_fields = ddb_deserializer.deserialize({'M': ddb[image_name]})
+
+        logger.debug('Deserialized doc_fields before GraphQL Schema mapping: %s', doc_fields)
+
+        # Map python types to match GrahpQL schema types
+        doc_fields = map_to_gql_types(doc_fields, doc_opensearch_index_name)
 
         # Sync enabled APIs do soft delete. We need to delete the record in OpenSearch if _deleted field is set
         if OPENSEARCH_USE_EXTERNAL_VERSIONING and event_name == 'MODIFY' and '_deleted' in  doc_fields and doc_fields['_deleted']:
@@ -259,4 +284,4 @@ def lambda_handler(event, context):
     try:
         return _lambda_handler(event, context)
     except Exception:
-        logger.error(traceback.format_exc())
+        logger.error(traceback.format_exc())