Map<String, DocumentField> readFields(String[] externalNames,
                                          String dataConnectionName,
                                          Map<String, String> options,
                                          boolean stream) {
        String collectionName = externalNames.length == 2 ? externalNames[1] : externalNames[0];
        String databaseName = Options.getDatabaseName(nodeEngine, externalNames, dataConnectionName);

        Map<String, DocumentField> fields = new LinkedHashMap<>();
        try (MongoClient client = connect(dataConnectionName, options)) {
            requireNonNull(client);

            ResourceChecks resourceChecks = readExistenceChecksFlag(options);
            if (resourceChecks.isEverPerformed()) {
                checkDatabaseAndCollectionExists(client, databaseName, collectionName);
            }

            MongoDatabase database = client.getDatabase(databaseName);
            List<Document> collections = database.listCollections()
                                                 .filter(eq("name", collectionName))
                                                 .into(new ArrayList<>());
            if (collections.isEmpty()) {
                ArrayList<String> list = database.listCollectionNames().into(new ArrayList<>());
                throw new IllegalArgumentException("collection " + collectionName + " was not found, maybe you mean: "
                        + list);
            }
            Document collectionInfo = collections.get(0);
            Document properties = getIgnoringNulls(collectionInfo, "options", "validator", "$jsonSchema", "properties");
            if (properties != null) {
                for (Entry<String, Object> property : properties.entrySet()) {
                    Document props = (Document) property.getValue();
                    BsonType bsonType = getBsonType(props);

                    String key = property.getKey();
                    if (stream) {
                        key = "fullDocument." + key;
                    }
                    fields.put(key, new DocumentField(bsonType, key));
                }
            } else {
                // fall back to sampling
                ArrayList<Document> samples =
                        database.getCollection(collectionName).find().limit(1).into(new ArrayList<>());
                if (samples.isEmpty()) {
                    throw new IllegalStateException("Cannot infer schema of collection " + collectionName
                            + ", no documents found");
                }
                Document sample = samples.get(0);
                for (Entry<String, Object> entry : sample.entrySet()) {
                    if (entry.getValue() == null) {
                        continue;
                    }
                    String key = entry.getKey();
                    if (stream) {
                        key = "fullDocument." + key;
                    }
                    DocumentField field = new DocumentField(resolveTypeFromJava(entry.getValue()), key);
                    fields.put(key, field);
                }
            }
            if (stream) {
                fields.put("operationType", new DocumentField(BsonType.STRING, "operationType"));
                fields.put("resumeToken", new DocumentField(BsonType.STRING, "resumeToken"));
                fields.put("wallTime", new DocumentField(BsonType.DATE_TIME, "wallTime"));
                fields.put("ts", new DocumentField(BsonType.TIMESTAMP, "ts"));
                fields.put("clusterTime", new DocumentField(BsonType.TIMESTAMP, "clusterTime"));
            }
        }
        return fields;
    }
@Test
    public void testResolvesFieldsViaSchema() {
        try (MongoClient client = MongoClients.create(mongoContainer.getConnectionString())) {
            String databaseName = "testDatabase";
            String collectionName = "people";
            MongoDatabase testDatabase = client.getDatabase(databaseName);

            CreateCollectionOptions options = new CreateCollectionOptions();
            ValidationOptions validationOptions = new ValidationOptions();
            validationOptions.validator(BsonDocument.parse(
                    """
                            {
                                $jsonSchema: {
                                  bsonType: "object",
                                  title: "Person Object Validation",
                                  required: [ "firstName", "lastName", "birthYear" ],
                                  properties: {\
                                    "firstName": { "bsonType": "string" }
                                    "lastName": { "bsonType": "string" }
                                    "birthYear": { "bsonType": "int" }
                                    "title": { "enum": [ "Bsc", "Msc", "PhD" ] }
                                    "intOrString": { "enum": [ "String", 1 ] }
                                    "unionType": { "bsonType": [ 'int', 'string' ] }
                                  }
                                }
                              }
                            """
            ));
            options.validationOptions(validationOptions);
            testDatabase.createCollection(collectionName, options);

            FieldResolver resolver = new FieldResolver(null);

            Map<String, String> readOpts = new HashMap<>();
            readOpts.put("connectionString", mongoContainer.getConnectionString());
            readOpts.put("database", databaseName);
            Map<String, DocumentField> fields = resolver.readFields(new String[]{databaseName, collectionName}, null,
                    readOpts, false);
            assertThat(fields)
                    .containsOnlyKeys("firstName", "lastName", "birthYear", "title", "unionType", "intOrString");
            assertThat(fields.get("lastName").columnType).isEqualTo(BsonType.STRING);
            assertThat(fields.get("birthYear").columnType).isEqualTo(BsonType.INT32);
            assertThat(fields.get("title").columnType).isEqualTo(BsonType.STRING);
            assertThat(fields.get("intOrString").columnType).isEqualTo(BsonType.DOCUMENT);
            assertThat(fields.get("unionType").columnType).isEqualTo(BsonType.DOCUMENT);
        }
    }