def METHOD_NAME(client: OpenSearch, record: Any) -> None:
    """
    Insert a new doc or modify an existing document. The latter happens when the workflow job is
    updated (new step, finishing). A record from torchci-workflow-job looks as follows
        {
            "eventID": "...",
            "eventName": "MODIFY",
            "eventVersion": "1.1",
            "eventSource": "aws:dynamodb",
            "awsRegion": "us-east-1",
            "dynamodb": {
                "ApproximateCreationDateTime": 1691722869,
                "Keys": {
                    "dynamoKey": {
                        "S": "pytorch/pytorch/15806159447"
                    }
                },
                "NewImage": {
                    "workflow_name": {
                        "S": "pull"
                    },
                    "steps": {
                        "L": [
                            {
                                "M": {
                                    "conclusion": {
                                        "NULL": True
                                    },
                                    "number": {
                                        "N": "1"
                                    },
                                    "completed_at": {
                                        "NULL": True
                                    },
                                    "name": {
                                        "S": "Set up job"
                                    },
                                    "started_at": {
                                        "S": "..."
                                    },
                                    "status": {
                                        "S": "in_progress"
                                    }
                                }
                            }
                         ]
                    },
                    ... all other fields ...
                },
                "OldImage": {
                    "workflow_name": {
                        "S": "pull"
                    },
                    "steps": {
                        "L": []
                    },
                    ... all other fields ...
                },
                "SequenceNumber": "...",
                "SizeBytes": 1763,
                "StreamViewType": "NEW_AND_OLD_IMAGES"
            },
            "eventSourceARN": "arn:aws:dynamodb:us-east-1:...:table/torchci-workflow-job/stream/..."
        }
    """
    index = extract_dynamodb_table(record)
    if not index:
        return
    # Create index using the table name if it's not there yet
    if not client.indices.exists(index):
        # https://www.elastic.co/guide/en/elasticsearch/reference/current/coerce.html
        client.indices.create(index, body={"settings": {"index.mapping.coerce": True}})
    body = unmarshal({"M": record.get("dynamodb", {}).get("NewImage", {})})
    if not body:
        return
    id = extract_dynamodb_key(record)
    if not id:
        return
    print(f"UPSERTING {id} INTO {index}")
    client.index(index=index, body=body, id=id, refresh=True)