public Option<Dataset<Row>> loadAsDataset(SparkSession spark, List<CloudObjectMetadata> cloudObjectMetadata,
                                            String fileFormat, Option<SchemaProvider> schemaProviderOption, int numPartitions) {
    if (LOG.isDebugEnabled()) {
      LOG.debug("Extracted distinct files " + cloudObjectMetadata.size()
          + " and some samples " + cloudObjectMetadata.stream().map(CloudObjectMetadata::getPath).limit(10).collect(Collectors.toList()));
    }

    if (isNullOrEmpty(cloudObjectMetadata)) {
      return Option.empty();
    }
    DataFrameReader reader = spark.read().format(fileFormat);
    String datasourceOpts = getStringWithAltKeys(properties, CloudSourceConfig.SPARK_DATASOURCE_OPTIONS, true);
    if (schemaProviderOption.isPresent()) {
      Schema sourceSchema = schemaProviderOption.get().getSourceSchema();
      if (sourceSchema != null && !sourceSchema.equals(InputBatch.NULL_SCHEMA)) {
        reader = reader.schema(AvroConversionUtils.convertAvroSchemaToStructType(sourceSchema));
      }
    }
    if (StringUtils.isNullOrEmpty(datasourceOpts)) {
      // fall back to legacy config for BWC. TODO consolidate in HUDI-6020
      datasourceOpts = getStringWithAltKeys(properties, S3EventsHoodieIncrSourceConfig.SPARK_DATASOURCE_OPTIONS, true);
    }
    if (StringUtils.nonEmpty(datasourceOpts)) {
      final ObjectMapper mapper = new ObjectMapper();
      Map<String, String> sparkOptionsMap = null;
      try {
        sparkOptionsMap = mapper.readValue(datasourceOpts, Map.class);
      } catch (IOException e) {
        throw new HoodieException(String.format("Failed to parse sparkOptions: %s", datasourceOpts), e);
      }
      LOG.info(String.format("sparkOptions loaded: %s", sparkOptionsMap));
      reader = reader.options(sparkOptionsMap);
    }
    List<String> paths = new ArrayList<>();
    for (CloudObjectMetadata o : cloudObjectMetadata) {
      paths.add(o.getPath());
    }
    boolean isCommaSeparatedPathFormat = properties.getBoolean(SPARK_DATASOURCE_READER_COMMA_SEPARATED_PATH_FORMAT.key(), false);

    Dataset<Row> dataset;
    if (isCommaSeparatedPathFormat) {
      dataset = reader.load(String.join(",", paths));
    } else {
      dataset = reader.load(paths.toArray(new String[cloudObjectMetadata.size()]));
    }

    // add partition column from source path if configured
    if (containsConfigProperty(properties, PATH_BASED_PARTITION_FIELDS)) {
      String[] partitionKeysToAdd = getStringWithAltKeys(properties, PATH_BASED_PARTITION_FIELDS).split(",");
      // Add partition column for all path-based partition keys. If key is not present in path, the value will be null.
      for (String partitionKey : partitionKeysToAdd) {
        String partitionPathPattern = String.format("%s=", partitionKey);
        LOG.info(String.format("Adding column %s to dataset", partitionKey));
        dataset = dataset.withColumn(partitionKey, split(split(input_file_name(), partitionPathPattern).getItem(1), "/").getItem(0));
      }
    }
    dataset = coalesceOrRepartition(dataset, numPartitions);
    return Option.of(dataset);
  }
@Test
  public void emptyMetadataReturnsEmptyOption() {
    CloudObjectsSelectorCommon cloudObjectsSelectorCommon = new CloudObjectsSelectorCommon(new TypedProperties());
    Option<Dataset<Row>> result = cloudObjectsSelectorCommon.loadAsDataset(sparkSession, Collections.emptyList(), "json", Option.empty(), 1);
    Assertions.assertFalse(result.isPresent());
  }