diff --git a/java/vortex-spark/src/main/java/dev/vortex/spark/VortexDataSourceV2.java b/java/vortex-spark/src/main/java/dev/vortex/spark/VortexDataSourceV2.java index bd434bcc8d2..74c8a8f4fc4 100644 --- a/java/vortex-spark/src/main/java/dev/vortex/spark/VortexDataSourceV2.java +++ b/java/vortex-spark/src/main/java/dev/vortex/spark/VortexDataSourceV2.java @@ -83,7 +83,8 @@ public StructType inferSchema(CaseInsensitiveStringMap options) { // If the path is a directory, scan the directory for a file and use that file if (!pathToInfer.endsWith(".vortex")) { Optional firstFile = - NativeFiles.listFiles(VortexSparkSession.get(formatOptions), pathToInfer, formatOptions).stream() + NativeFiles.listFiles(VortexSparkSession.get(formatOptions), pathToInfer, formatOptions.asMap()) + .stream() .findFirst(); if (firstFile.isEmpty()) { @@ -99,7 +100,7 @@ public StructType inferSchema(CaseInsensitiveStringMap options) { StructType dataSchema; { - DataSource ds = DataSource.open(VortexSparkSession.get(formatOptions), pathToInfer, formatOptions); + DataSource ds = DataSource.open(VortexSparkSession.get(formatOptions), pathToInfer, formatOptions.asMap()); var arrowSchema = ds.arrowSchema(dev.vortex.arrow.ArrowAllocation.rootAllocator()); StructField[] fields = arrowSchema.getFields().stream() .map(f -> new StructField( @@ -145,7 +146,8 @@ public Transform[] inferPartitioning(CaseInsensitiveStringMap options) { String pathToInfer = Objects.requireNonNull(Iterables.getLast(paths)); if (!pathToInfer.endsWith(".vortex")) { Optional firstFile = - NativeFiles.listFiles(VortexSparkSession.get(formatOptions), pathToInfer, formatOptions).stream() + NativeFiles.listFiles(VortexSparkSession.get(formatOptions), pathToInfer, formatOptions.asMap()) + .stream() .findFirst(); if (firstFile.isEmpty()) { return new Transform[0]; @@ -208,7 +210,7 @@ public String shortName() { return "vortex"; } - private Map buildDataSourceOptions(Map properties) { + private VortexOptions buildDataSourceOptions(Map properties) { var hadoopConf = sparkSession.get().sessionState().newHadoopConf(); var options = ImmutableMap.builder(); @@ -219,7 +221,7 @@ private Map buildDataSourceOptions(Map propertie // Forward any Azure-relevant properties from hadoopConf to the reader config. options.putAll(HadoopUtils.azurePropertiesFromHadoopConf(hadoopConf)); - return options.build(); + return VortexOptions.of(options.build()); } private static ImmutableList getPathsOrEmpty(CaseInsensitiveStringMap uncased) { diff --git a/java/vortex-spark/src/main/java/dev/vortex/spark/VortexFilePartition.java b/java/vortex-spark/src/main/java/dev/vortex/spark/VortexFilePartition.java index 4247788a887..031f07ea917 100644 --- a/java/vortex-spark/src/main/java/dev/vortex/spark/VortexFilePartition.java +++ b/java/vortex-spark/src/main/java/dev/vortex/spark/VortexFilePartition.java @@ -25,8 +25,5 @@ * @param partitionValues Hive-style partition column values shared by all {@link #paths()} */ public record VortexFilePartition( - List paths, - StructType readSchema, - Map formatOptions, - Map partitionValues) + List paths, StructType readSchema, VortexOptions formatOptions, Map partitionValues) implements InputPartition, Serializable {} diff --git a/java/vortex-spark/src/main/java/dev/vortex/spark/VortexOptions.java b/java/vortex-spark/src/main/java/dev/vortex/spark/VortexOptions.java new file mode 100644 index 00000000000..ca5ff030d56 --- /dev/null +++ b/java/vortex-spark/src/main/java/dev/vortex/spark/VortexOptions.java @@ -0,0 +1,203 @@ +// SPDX-License-Identifier: Apache-2.0 +// SPDX-FileCopyrightText: Copyright the Vortex contributors + +package dev.vortex.spark; + +import com.google.common.collect.ImmutableMap; +import java.io.Serializable; +import java.util.HashSet; +import java.util.LinkedHashMap; +import java.util.Locale; +import java.util.Map; +import java.util.Objects; +import java.util.Optional; +import java.util.Set; +import org.apache.spark.sql.util.CaseInsensitiveStringMap; + +/** + * The Vortex format options of a read or write, resolved case-insensitively. + * + *

Spark lower-cases the keys of the {@link CaseInsensitiveStringMap} it hands to a table, so an option a user + * spelled {@code vortex.workerThreads} arrives as {@code vortex.workerthreads}. Holding the options here means every + * call site gets that matching for free, and each option has one place that names it, documents its default and + * validates it — following the shape of Spark's own {@code ParquetOptions}. + * + *

Instances cross Spark's serialization boundary to the executors, so the case-insensitive view is {@code transient} + * and rebuilt on demand from the raw map. + */ +public final class VortexOptions implements Serializable { + private static final long serialVersionUID = 1L; + + /** Number of native worker threads used to decode a scan. */ + public static final String WORKER_THREADS = "vortex.workerThreads"; + + /** Rows buffered before a batch is written out. */ + public static final String WRITE_BATCH_SIZE = "vortex.write.batch.size"; + + /** Legacy spelling of {@link #WRITE_BATCH_SIZE}, still honoured. */ + public static final String LEGACY_WRITE_BATCH_SIZE = "batch.size"; + + /** Class name of a {@link VortexSessionProvider} supplying the native session. */ + public static final String SESSION_PROVIDER = "vortex.session.provider"; + + /** Default number of native worker threads. */ + public static final int DEFAULT_WORKER_THREADS = 4; + + /** Default number of rows buffered before a write batch is flushed. */ + public static final int DEFAULT_WRITE_BATCH_SIZE = 2048; + + /** Smallest accepted {@link #WRITE_BATCH_SIZE}. */ + public static final int MIN_WRITE_BATCH_SIZE = 1; + + /** Largest accepted {@link #WRITE_BATCH_SIZE}. */ + public static final int MAX_WRITE_BATCH_SIZE = 65536; + + private final ImmutableMap options; + + private transient CaseInsensitiveStringMap cached; + + private VortexOptions(Map options) { + this.options = ImmutableMap.copyOf(options); + } + + /** Wraps the supplied options; the map is copied, so later changes to it are not observed. */ + public static VortexOptions of(Map options) { + return new VortexOptions(Objects.requireNonNull(options, "options")); + } + + /** Empty options, for reads and writes that configure nothing. */ + public static VortexOptions empty() { + return new VortexOptions(ImmutableMap.of()); + } + + /** + * Returns these options with {@code overrides} applied on top, matching keys case-insensitively so that an override + * replaces the option it means to rather than sitting beside it under a different spelling. + */ + public VortexOptions withOverrides(Map overrides) { + if (overrides.isEmpty()) { + return this; + } + Set overridden = new HashSet<>(); + overrides.keySet().forEach(key -> overridden.add(fold(key))); + Map merged = new LinkedHashMap<>(); + options.forEach((key, value) -> { + if (!overridden.contains(fold(key))) { + merged.put(key, value); + } + }); + merged.putAll(overrides); + return new VortexOptions(merged); + } + + /** + * Number of native worker threads to decode with, {@value #DEFAULT_WORKER_THREADS} if unset. + * + * @throws IllegalArgumentException if the value is not a non-negative integer + */ + public int workerThreads() { + int threads = intOption(WORKER_THREADS, DEFAULT_WORKER_THREADS); + if (threads < 0) { + throw new IllegalArgumentException( + String.format("%s must be a non-negative integer, got %d", WORKER_THREADS, threads)); + } + return threads; + } + + /** + * Rows to buffer before writing a batch, {@value #DEFAULT_WRITE_BATCH_SIZE} if unset. A value outside + * [{@value #MIN_WRITE_BATCH_SIZE}, {@value #MAX_WRITE_BATCH_SIZE}] falls back to the default; use + * {@link #rejectedWriteBatchSize()} to report which value was ignored. + * + * @throws IllegalArgumentException if the value is not an integer + */ + public int writeBatchSize() { + int configured = configuredWriteBatchSize(); + if (configured < MIN_WRITE_BATCH_SIZE || configured > MAX_WRITE_BATCH_SIZE) { + return DEFAULT_WRITE_BATCH_SIZE; + } + return configured; + } + + /** + * The out-of-range batch size {@link #writeBatchSize()} had to ignore, together with the key it was set under, or + * empty when the configured value was usable. + */ + public Optional rejectedWriteBatchSize() { + int configured = configuredWriteBatchSize(); + if (configured >= MIN_WRITE_BATCH_SIZE && configured <= MAX_WRITE_BATCH_SIZE) { + return Optional.empty(); + } + String key = caseInsensitive().get(WRITE_BATCH_SIZE) == null ? LEGACY_WRITE_BATCH_SIZE : WRITE_BATCH_SIZE; + return Optional.of(new RejectedOption(key, configured)); + } + + /** An option value that was parsed but fell outside its accepted range. */ + public record RejectedOption(String key, int value) {} + + private int configuredWriteBatchSize() { + Integer current = optionalIntOption(WRITE_BATCH_SIZE); + if (current != null) { + return current; + } + Integer legacy = optionalIntOption(LEGACY_WRITE_BATCH_SIZE); + return legacy != null ? legacy : DEFAULT_WRITE_BATCH_SIZE; + } + + /** Class name of the session provider to use, empty to use the default session. */ + public Optional sessionProvider() { + String provider = caseInsensitive().get(SESSION_PROVIDER); + return provider == null || provider.isEmpty() ? Optional.empty() : Optional.of(provider); + } + + /** The raw options, as the native bindings expect them. */ + public Map asMap() { + return options; + } + + @Override + public boolean equals(Object other) { + return other instanceof VortexOptions && options.equals(((VortexOptions) other).options); + } + + @Override + public int hashCode() { + return options.hashCode(); + } + + @Override + public String toString() { + return options.toString(); + } + + private int intOption(String key, int defaultValue) { + Integer value = optionalIntOption(key); + return value != null ? value : defaultValue; + } + + private Integer optionalIntOption(String key) { + String value = caseInsensitive().get(key); + if (value == null) { + return null; + } + try { + return Integer.valueOf(value.trim()); + } catch (NumberFormatException e) { + throw new IllegalArgumentException(String.format("%s must be an integer, got \"%s\"", key, value), e); + } + } + + private CaseInsensitiveStringMap caseInsensitive() { + CaseInsensitiveStringMap local = cached; + if (local == null) { + local = new CaseInsensitiveStringMap(options); + cached = local; + } + return local; + } + + /** Folds a key the same way {@link CaseInsensitiveStringMap} does, so the two agree on what collides. */ + private static String fold(String key) { + return key.toLowerCase(Locale.ROOT); + } +} diff --git a/java/vortex-spark/src/main/java/dev/vortex/spark/VortexSparkSession.java b/java/vortex-spark/src/main/java/dev/vortex/spark/VortexSparkSession.java index 1d3d07d3b25..3f2054d42ec 100644 --- a/java/vortex-spark/src/main/java/dev/vortex/spark/VortexSparkSession.java +++ b/java/vortex-spark/src/main/java/dev/vortex/spark/VortexSparkSession.java @@ -4,8 +4,8 @@ package dev.vortex.spark; import dev.vortex.api.Session; -import java.util.Map; import java.util.Objects; +import java.util.Optional; import java.util.concurrent.ConcurrentHashMap; /** @@ -31,7 +31,7 @@ */ public final class VortexSparkSession { /** Options key used to select a {@link VortexSessionProvider} by class name. */ - public static final String PROVIDER_OPTION = "vortex.session.provider"; + public static final String PROVIDER_OPTION = VortexOptions.SESSION_PROVIDER; private static final ConcurrentHashMap providerCache = new ConcurrentHashMap<>(); private static volatile Session defaultSession; @@ -53,15 +53,14 @@ public static Session get() { } /** - * Resolve the session to use for a given set of Spark format options. Honours the {@value #PROVIDER_OPTION} key; + * Resolve the session to use for a given set of Vortex format options. Honours the {@value #PROVIDER_OPTION} key; * falls back to {@link #get()} otherwise. */ - public static Session get(Map options) { - String providerClass = options == null ? null : options.get(PROVIDER_OPTION); - if (providerClass == null || providerClass.isEmpty()) { - return get(); - } - return providerCache.computeIfAbsent(providerClass, VortexSparkSession::loadProvider); + public static Session get(VortexOptions options) { + Optional providerClass = options == null ? Optional.empty() : options.sessionProvider(); + return providerClass + .map(className -> providerCache.computeIfAbsent(className, VortexSparkSession::loadProvider)) + .orElseGet(VortexSparkSession::get); } /** diff --git a/java/vortex-spark/src/main/java/dev/vortex/spark/VortexTable.java b/java/vortex-spark/src/main/java/dev/vortex/spark/VortexTable.java index f65f74ccf19..da07e810d4e 100644 --- a/java/vortex-spark/src/main/java/dev/vortex/spark/VortexTable.java +++ b/java/vortex-spark/src/main/java/dev/vortex/spark/VortexTable.java @@ -6,11 +6,9 @@ import com.google.common.collect.ImmutableList; import com.google.common.collect.ImmutableSet; import com.google.common.collect.Iterables; -import com.google.common.collect.Maps; import dev.vortex.spark.read.VortexScanBuilder; import dev.vortex.spark.write.VortexWriteBuilder; import java.util.Arrays; -import java.util.Map; import java.util.Set; import org.apache.spark.sql.connector.catalog.CatalogV2Util; import org.apache.spark.sql.connector.catalog.SupportsRead; @@ -30,14 +28,14 @@ public final class VortexTable implements Table, SupportsRead, SupportsWrite { private final ImmutableList paths; private final StructType schema; - private final Map formatOptions; + private final VortexOptions formatOptions; private final Transform[] partitionTransforms; /** Creates a new VortexTable with read/write support. */ public VortexTable( ImmutableList paths, StructType schema, - Map formatOptions, + VortexOptions formatOptions, Transform[] partitionTransforms) { this.paths = paths; this.schema = schema; @@ -55,10 +53,7 @@ public VortexTable( */ @Override public ScanBuilder newScanBuilder(CaseInsensitiveStringMap options) { - Map opts = Maps.newHashMap(); - opts.putAll(formatOptions); - opts.putAll(options); - return new VortexScanBuilder(opts, partitionTransforms) + return new VortexScanBuilder(formatOptions.withOverrides(options), partitionTransforms) .addAllPaths(paths) .addAllColumns(Arrays.asList(CatalogV2Util.structTypeToV2Columns(schema))); } diff --git a/java/vortex-spark/src/main/java/dev/vortex/spark/read/VortexBatchExec.java b/java/vortex-spark/src/main/java/dev/vortex/spark/read/VortexBatchExec.java index 198dfd6c77c..e7a7ed115ab 100644 --- a/java/vortex-spark/src/main/java/dev/vortex/spark/read/VortexBatchExec.java +++ b/java/vortex-spark/src/main/java/dev/vortex/spark/read/VortexBatchExec.java @@ -7,11 +7,13 @@ import dev.vortex.api.Session; import dev.vortex.jni.NativeFiles; import dev.vortex.spark.VortexFilePartition; +import dev.vortex.spark.VortexOptions; import dev.vortex.spark.VortexSparkSession; import java.util.Arrays; import java.util.HashSet; import java.util.List; import java.util.Map; +import java.util.Objects; import java.util.Set; import java.util.stream.Collectors; import java.util.stream.Stream; @@ -27,7 +29,7 @@ public final class VortexBatchExec implements Batch { private final List paths; private final StructType readSchema; - private final Map formatOptions; + private final VortexOptions formatOptions; private final Predicate[] pushedPredicates; private List resolvedPaths; @@ -40,10 +42,10 @@ public final class VortexBatchExec implements Batch { * time */ public VortexBatchExec( - List paths, List columns, Map formatOptions, Predicate[] pushedPredicates) { + List paths, List columns, VortexOptions formatOptions, Predicate[] pushedPredicates) { this.paths = List.copyOf(paths); this.readSchema = CatalogV2Util.v2ColumnsToStructType(columns.toArray(new Column[0])); - this.formatOptions = Map.copyOf(formatOptions); + this.formatOptions = Objects.requireNonNull(formatOptions, "formatOptions"); this.pushedPredicates = pushedPredicates == null ? new Predicate[0] : pushedPredicates.clone(); } @@ -68,6 +70,9 @@ public InputPartition[] planInputPartitions() { @Override public PartitionReaderFactory createReaderFactory() { + // Resolve the worker-thread count here, on the driver: an invalid value then fails the query + // once, rather than once per task after the readers have already been shipped to executors. + formatOptions.workerThreads(); List files = resolvedPaths != null ? resolvedPaths : resolvePaths(); Set partitionColumns = collectPartitionColumnNames(files); List dataColumnNames = Arrays.stream(readSchema.fieldNames()) @@ -85,11 +90,11 @@ private List resolvePaths() { * file are kept as-is. Shared with {@link VortexScan#estimateStatistics()} so planning and execution resolve paths * identically. */ - static List resolveVortexPaths(Session session, List paths, Map formatOptions) { + static List resolveVortexPaths(Session session, List paths, VortexOptions formatOptions) { return paths.stream() .flatMap(path -> path.endsWith(".vortex") ? Stream.of(path) - : NativeFiles.listFiles(session, path, formatOptions).stream()) + : NativeFiles.listFiles(session, path, formatOptions.asMap()).stream()) .collect(Collectors.toList()); } diff --git a/java/vortex-spark/src/main/java/dev/vortex/spark/read/VortexPartitionReader.java b/java/vortex-spark/src/main/java/dev/vortex/spark/read/VortexPartitionReader.java index f9cd2363d59..f9dfd724138 100644 --- a/java/vortex-spark/src/main/java/dev/vortex/spark/read/VortexPartitionReader.java +++ b/java/vortex-spark/src/main/java/dev/vortex/spark/read/VortexPartitionReader.java @@ -14,6 +14,7 @@ import dev.vortex.relocated.org.apache.arrow.vector.VectorSchemaRoot; import dev.vortex.relocated.org.apache.arrow.vector.ipc.ArrowReader; import dev.vortex.spark.VortexFilePartition; +import dev.vortex.spark.VortexOptions; import dev.vortex.spark.VortexSparkSession; import java.io.IOException; import java.util.List; @@ -51,13 +52,13 @@ final class VortexPartitionReader implements PartitionReader { VortexPartitionReader( VortexFilePartition spark, List dataColumnNames, - Map formatOptions, + VortexOptions formatOptions, Predicate[] pushedPredicates) { this.spark = spark; this.allocator = ArrowAllocation.rootAllocator(); session = VortexSparkSession.get(formatOptions); - dataSource = DataSource.open(session, spark.paths(), formatOptions); + dataSource = DataSource.open(session, spark.paths(), formatOptions.asMap()); var options = ScanOptions.builder(); if (!dataColumnNames.isEmpty()) { diff --git a/java/vortex-spark/src/main/java/dev/vortex/spark/read/VortexPartitionReaderFactory.java b/java/vortex-spark/src/main/java/dev/vortex/spark/read/VortexPartitionReaderFactory.java index e187e4863b1..8013da8ff70 100644 --- a/java/vortex-spark/src/main/java/dev/vortex/spark/read/VortexPartitionReaderFactory.java +++ b/java/vortex-spark/src/main/java/dev/vortex/spark/read/VortexPartitionReaderFactory.java @@ -4,12 +4,12 @@ package dev.vortex.spark.read; import com.google.common.collect.ImmutableList; -import com.google.common.collect.ImmutableMap; import dev.vortex.jni.NativeRuntime; import dev.vortex.spark.VortexFilePartition; +import dev.vortex.spark.VortexOptions; import java.io.Serializable; import java.util.List; -import java.util.Map; +import java.util.Objects; import org.apache.spark.sql.catalyst.InternalRow; import org.apache.spark.sql.connector.expressions.filter.Predicate; import org.apache.spark.sql.connector.read.InputPartition; @@ -29,13 +29,13 @@ public final class VortexPartitionReaderFactory implements PartitionReaderFactor private static final long serialVersionUID = 1L; private final ImmutableList dataColumnNames; - private final ImmutableMap formatOptions; + private final VortexOptions formatOptions; private final Predicate[] pushedPredicates; public VortexPartitionReaderFactory( - List dataColumnNames, Map formatOptions, Predicate[] pushedPredicates) { + List dataColumnNames, VortexOptions formatOptions, Predicate[] pushedPredicates) { this.dataColumnNames = ImmutableList.copyOf(dataColumnNames); - this.formatOptions = ImmutableMap.copyOf(formatOptions); + this.formatOptions = Objects.requireNonNull(formatOptions, "formatOptions"); this.pushedPredicates = pushedPredicates == null ? new Predicate[0] : pushedPredicates.clone(); } @@ -46,7 +46,7 @@ public PartitionReader createReader(InputPartition partition) { @Override public PartitionReader createColumnarReader(InputPartition partition) { - NativeRuntime.setWorkerThreads(Integer.parseInt(formatOptions.getOrDefault("vortex.workerThreads", "4"))); + NativeRuntime.setWorkerThreads(formatOptions.workerThreads()); VortexFilePartition spark = (VortexFilePartition) partition; return new VortexPartitionReader(spark, dataColumnNames, formatOptions, pushedPredicates); } diff --git a/java/vortex-spark/src/main/java/dev/vortex/spark/read/VortexScan.java b/java/vortex-spark/src/main/java/dev/vortex/spark/read/VortexScan.java index 02a7563f925..0c100d83458 100644 --- a/java/vortex-spark/src/main/java/dev/vortex/spark/read/VortexScan.java +++ b/java/vortex-spark/src/main/java/dev/vortex/spark/read/VortexScan.java @@ -5,6 +5,7 @@ import dev.vortex.api.DataSource; import dev.vortex.api.Session; +import dev.vortex.spark.VortexOptions; import dev.vortex.spark.VortexSparkSession; import java.util.Arrays; import java.util.List; @@ -37,7 +38,7 @@ public final class VortexScan implements Scan, SupportsReportStatistics { private final List paths; private final List tableColumns; private final List readColumns; - private final Map formatOptions; + private final VortexOptions formatOptions; private final Predicate[] pushedPredicates; private volatile Statistics cachedStatistics; @@ -56,7 +57,7 @@ public VortexScan( List tableColumns, List readColumns, Predicate[] pushedPredicates, - Map formatOptions) { + VortexOptions formatOptions) { this.paths = paths; this.tableColumns = tableColumns; this.readColumns = readColumns; @@ -141,7 +142,7 @@ private Statistics computeStatistics() { return new VortexStatistics(OptionalLong.empty(), OptionalLong.empty()); } - DataSource source = DataSource.open(session, resolvedPaths, formatOptions); + DataSource source = DataSource.open(session, resolvedPaths, formatOptions.asMap()); return new VortexStatistics( source.rowCount().asOptional(), scaleSizeInBytes(source.byteSize().asOptional())); diff --git a/java/vortex-spark/src/main/java/dev/vortex/spark/read/VortexScanBuilder.java b/java/vortex-spark/src/main/java/dev/vortex/spark/read/VortexScanBuilder.java index 62c8085aa0f..8a42dc4b827 100644 --- a/java/vortex-spark/src/main/java/dev/vortex/spark/read/VortexScanBuilder.java +++ b/java/vortex-spark/src/main/java/dev/vortex/spark/read/VortexScanBuilder.java @@ -6,7 +6,7 @@ import static com.google.common.base.Preconditions.checkState; import com.google.common.collect.ImmutableList; -import com.google.common.collect.Maps; +import dev.vortex.spark.VortexOptions; import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; @@ -33,12 +33,12 @@ public final class VortexScanBuilder private final ImmutableList.Builder paths; private final List tableColumns; private final List readColumns; - private final Map formatOptions; + private final VortexOptions formatOptions; private final Set partitionColumnNames; private Predicate[] pushedPredicates = new Predicate[0]; /** Creates a new VortexScanBuilder with empty paths and columns. */ - public VortexScanBuilder(Map formatOptions) { + public VortexScanBuilder(VortexOptions formatOptions) { this(formatOptions, new Transform[0]); } @@ -47,14 +47,11 @@ public VortexScanBuilder(Map formatOptions) { * reference partition columns are not pushed down, since the partition columns are not stored inside the Vortex * files. */ - public VortexScanBuilder(Map formatOptions, Transform[] partitionTransforms) { + public VortexScanBuilder(VortexOptions formatOptions, Transform[] partitionTransforms) { this.paths = ImmutableList.builder(); - Map options = Maps.newHashMap(); - options.put("vortex.workerThreads", "4"); - options.putAll(formatOptions); this.tableColumns = new ArrayList<>(); this.readColumns = new ArrayList<>(); - this.formatOptions = options; + this.formatOptions = formatOptions; this.partitionColumnNames = collectPartitionColumnNames(partitionTransforms); } diff --git a/java/vortex-spark/src/main/java/dev/vortex/spark/write/PartitionedVortexDataWriter.java b/java/vortex-spark/src/main/java/dev/vortex/spark/write/PartitionedVortexDataWriter.java index 01ed570e4fc..fdf9ebde83e 100644 --- a/java/vortex-spark/src/main/java/dev/vortex/spark/write/PartitionedVortexDataWriter.java +++ b/java/vortex-spark/src/main/java/dev/vortex/spark/write/PartitionedVortexDataWriter.java @@ -5,6 +5,7 @@ import com.google.common.collect.ImmutableList; import com.google.common.primitives.ImmutableIntArray; +import dev.vortex.spark.VortexOptions; import java.io.IOException; import java.io.Serializable; import java.net.URLEncoder; @@ -46,7 +47,6 @@ import org.apache.spark.sql.types.StructType; import org.apache.spark.sql.types.TimestampNTZType; import org.apache.spark.sql.types.TimestampType; -import org.apache.spark.sql.util.CaseInsensitiveStringMap; import org.apache.spark.unsafe.types.UTF8String; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -65,7 +65,7 @@ public final class PartitionedVortexDataWriter implements DataWriter options; + private final VortexOptions options; private final boolean overwrite; // Resolved eagerly so that Spark Transform objects (Scala case classes that are not // Java-serializable) never reach the DataWriterFactory serialization boundary. @@ -50,7 +50,7 @@ public final class VortexBatchWrite implements Write, BatchWrite, Serializable { VortexBatchWrite( String outputPath, StructType schema, - Map options, + VortexOptions options, boolean overwrite, Transform[] partitionTransforms) { this.outputPath = outputPath; @@ -85,7 +85,7 @@ public DataWriterFactory createBatchWriterFactory(PhysicalWriteInfo info) { // Handle overwrite cleanup BEFORE writing starts if (overwrite) { var session = VortexSparkSession.get(options); - var uris = NativeFiles.listFiles(session, outputPath, options); + var uris = NativeFiles.listFiles(session, outputPath, options.asMap()); // Deleting the existing files is destructive and happens before the new data is written: // if the subsequent write fails, abort() only removes the newly written files and cannot // restore what was deleted here. Log loudly so operators can see what was removed. @@ -94,7 +94,7 @@ public DataWriterFactory createBatchWriterFactory(PhysicalWriteInfo info) { + "this cannot be undone if the subsequent write fails", uris.size(), outputPath); - NativeFiles.delete(session, uris.toArray(new String[0]), options); + NativeFiles.delete(session, uris.toArray(new String[0]), options.asMap()); } return new VortexDataWriterFactory(outputPath, schema, options, resolvedTransforms); @@ -146,7 +146,7 @@ public void abort(WriterCommitMessage[] messages) { } log.warn("Deleting {} file(s) written before the job failed, under {}", filePaths.size(), outputPath); try { - NativeFiles.delete(VortexSparkSession.get(options), filePaths.toArray(new String[0]), options); + NativeFiles.delete(VortexSparkSession.get(options), filePaths.toArray(new String[0]), options.asMap()); } catch (RuntimeException e) { log.error("Failed to clean up {} file(s) under {}", filePaths.size(), outputPath, e); } diff --git a/java/vortex-spark/src/main/java/dev/vortex/spark/write/VortexDataWriter.java b/java/vortex-spark/src/main/java/dev/vortex/spark/write/VortexDataWriter.java index dfadd320a43..f786e71612f 100644 --- a/java/vortex-spark/src/main/java/dev/vortex/spark/write/VortexDataWriter.java +++ b/java/vortex-spark/src/main/java/dev/vortex/spark/write/VortexDataWriter.java @@ -28,6 +28,7 @@ import dev.vortex.relocated.org.apache.arrow.vector.complex.ListVector; import dev.vortex.relocated.org.apache.arrow.vector.complex.MapVector; import dev.vortex.relocated.org.apache.arrow.vector.complex.StructVector; +import dev.vortex.spark.VortexOptions; import dev.vortex.spark.VortexSparkSession; import java.io.IOException; import java.nio.file.Files; @@ -58,7 +59,6 @@ import org.apache.spark.sql.types.StructType; import org.apache.spark.sql.types.TimestampNTZType; import org.apache.spark.sql.types.TimestampType; -import org.apache.spark.sql.util.CaseInsensitiveStringMap; import org.apache.spark.unsafe.types.UTF8String; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -72,13 +72,9 @@ public final class VortexDataWriter implements DataWriter, AutoCloseable { private static final Logger logger = LoggerFactory.getLogger(VortexDataWriter.class); - private static final int DEFAULT_BATCH_SIZE = 2048; - private static final int MIN_BATCH_SIZE = 1; - private static final int MAX_BATCH_SIZE = 65536; // 64K rows max per batch - private final String filePath; private final StructType schema; - private final CaseInsensitiveStringMap options; + private final VortexOptions options; private final int batchSize; private Session session; @@ -97,37 +93,34 @@ public final class VortexDataWriter implements DataWriter, AutoClos * @param schema the schema of the data to write * @param options additional write options */ - VortexDataWriter(String filePath, StructType schema, CaseInsensitiveStringMap options) { + VortexDataWriter(String filePath, StructType schema, VortexOptions options) { this.filePath = filePath; this.schema = schema; this.options = options; - // Get batch size from options with validation - // Users can set this with: .option("vortex.write.batch.size", "4096") - int configuredBatchSize = - options.getInt("vortex.write.batch.size", options.getInt("batch.size", DEFAULT_BATCH_SIZE)); - if (configuredBatchSize < MIN_BATCH_SIZE || configuredBatchSize > MAX_BATCH_SIZE) { - logger.warn( - "Batch size {} is out of valid range [{}, {}], using default: {}", - configuredBatchSize, - MIN_BATCH_SIZE, - MAX_BATCH_SIZE, - DEFAULT_BATCH_SIZE); - this.batchSize = DEFAULT_BATCH_SIZE; - } else { - this.batchSize = configuredBatchSize; - if (this.batchSize != DEFAULT_BATCH_SIZE) { - logger.debug("Using configured batch size: {}", this.batchSize); - } - } + this.batchSize = options.writeBatchSize(); + options.rejectedWriteBatchSize() + .ifPresentOrElse( + rejected -> logger.warn( + "{}={} is out of the valid range [{}, {}], using the default of {}", + rejected.key(), + rejected.value(), + VortexOptions.MIN_WRITE_BATCH_SIZE, + VortexOptions.MAX_WRITE_BATCH_SIZE, + VortexOptions.DEFAULT_WRITE_BATCH_SIZE), + () -> { + if (this.batchSize != VortexOptions.DEFAULT_WRITE_BATCH_SIZE) { + logger.debug("Using configured batch size: {}", this.batchSize); + } + }); try { this.allocator = new RootAllocator(); var arrowSchema = SparkToArrowSchema.convert(schema); - this.session = VortexSparkSession.get(options.asCaseSensitiveMap()); + this.session = VortexSparkSession.get(options); this.vortexWriter = VortexWriter.builder(session, filePath, arrowSchema, allocator) - .options(options.asCaseSensitiveMap()) + .options(options.asMap()) .build(); this.vectorSchemaRoot = VectorSchemaRoot.create(arrowSchema, allocator); diff --git a/java/vortex-spark/src/main/java/dev/vortex/spark/write/VortexDataWriterFactory.java b/java/vortex-spark/src/main/java/dev/vortex/spark/write/VortexDataWriterFactory.java index e8237bde309..a9ab3b20530 100644 --- a/java/vortex-spark/src/main/java/dev/vortex/spark/write/VortexDataWriterFactory.java +++ b/java/vortex-spark/src/main/java/dev/vortex/spark/write/VortexDataWriterFactory.java @@ -3,13 +3,12 @@ package dev.vortex.spark.write; +import dev.vortex.spark.VortexOptions; import java.io.Serializable; -import java.util.Map; import org.apache.spark.sql.catalyst.InternalRow; import org.apache.spark.sql.connector.write.DataWriter; import org.apache.spark.sql.connector.write.DataWriterFactory; import org.apache.spark.sql.types.StructType; -import org.apache.spark.sql.util.CaseInsensitiveStringMap; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -25,8 +24,7 @@ public final class VortexDataWriterFactory implements DataWriterFactory, Seriali private final String outputUri; private final StructType schema; - // Store options as a serializable Map instead of CaseInsensitiveStringMap - private final Map options; + private final VortexOptions options; private final PartitionedVortexDataWriter.ResolvedTransform[] resolvedTransforms; /** @@ -40,7 +38,7 @@ public final class VortexDataWriterFactory implements DataWriterFactory, Seriali VortexDataWriterFactory( String outputUri, StructType schema, - Map options, + VortexOptions options, PartitionedVortexDataWriter.ResolvedTransform[] resolvedTransforms) { this.outputUri = outputUri; this.schema = schema; @@ -62,12 +60,9 @@ public final class VortexDataWriterFactory implements DataWriterFactory, Seriali public DataWriter createWriter(int partitionId, long taskId) { log.debug("Creating writer for partition={} task={}", partitionId, taskId); - CaseInsensitiveStringMap optionsMap = new CaseInsensitiveStringMap(options); - if (resolvedTransforms.length > 0) { log.debug("Creating partitioned writer with {} transforms", resolvedTransforms.length); - return new PartitionedVortexDataWriter( - outputUri, schema, optionsMap, resolvedTransforms, partitionId, taskId); + return new PartitionedVortexDataWriter(outputUri, schema, options, resolvedTransforms, partitionId, taskId); } // Non-partitioned write: single file per task @@ -80,6 +75,6 @@ public DataWriter createWriter(int partitionId, long taskId) { } log.debug("Output file: {}", fileUri); - return new VortexDataWriter(fileUri, schema, optionsMap); + return new VortexDataWriter(fileUri, schema, options); } } diff --git a/java/vortex-spark/src/main/java/dev/vortex/spark/write/VortexWriteBuilder.java b/java/vortex-spark/src/main/java/dev/vortex/spark/write/VortexWriteBuilder.java index 921e586a910..ac2a54e2ea6 100644 --- a/java/vortex-spark/src/main/java/dev/vortex/spark/write/VortexWriteBuilder.java +++ b/java/vortex-spark/src/main/java/dev/vortex/spark/write/VortexWriteBuilder.java @@ -3,7 +3,7 @@ package dev.vortex.spark.write; -import java.util.Map; +import dev.vortex.spark.VortexOptions; import org.apache.spark.sql.connector.expressions.Transform; import org.apache.spark.sql.connector.write.LogicalWriteInfo; import org.apache.spark.sql.connector.write.SupportsTruncate; @@ -20,7 +20,7 @@ public final class VortexWriteBuilder implements WriteBuilder, SupportsTruncate private final String paths; private final LogicalWriteInfo writeInfo; - private final Map options; + private final VortexOptions options; private final Transform[] partitionTransforms; private boolean truncate = false; @@ -33,7 +33,7 @@ public final class VortexWriteBuilder implements WriteBuilder, SupportsTruncate * @param partitionTransforms partition transforms (may be empty) */ public VortexWriteBuilder( - String paths, LogicalWriteInfo writeInfo, Map options, Transform[] partitionTransforms) { + String paths, LogicalWriteInfo writeInfo, VortexOptions options, Transform[] partitionTransforms) { this.paths = paths; this.writeInfo = writeInfo; this.options = options; diff --git a/java/vortex-spark/src/test/java/dev/vortex/spark/VortexDataSourceStatsTest.java b/java/vortex-spark/src/test/java/dev/vortex/spark/VortexDataSourceStatsTest.java index 0595349a49a..e59eb2fced9 100644 --- a/java/vortex-spark/src/test/java/dev/vortex/spark/VortexDataSourceStatsTest.java +++ b/java/vortex-spark/src/test/java/dev/vortex/spark/VortexDataSourceStatsTest.java @@ -13,7 +13,6 @@ import java.nio.file.Files; import java.nio.file.Path; import java.util.Comparator; -import java.util.Map; import java.util.stream.Stream; import org.apache.spark.sql.Dataset; import org.apache.spark.sql.Row; @@ -180,7 +179,7 @@ private VortexScan buildScan(Path outputPath, StructType requiredSchema) { .load(); StructType readSchema = readDf.schema(); - VortexScanBuilder builder = new VortexScanBuilder(Map.of()); + VortexScanBuilder builder = new VortexScanBuilder(VortexOptions.empty()); builder.addPath(outputPath.toUri().toString()); for (StructField field : readSchema.fields()) { builder.addColumn(Column.create(field.name(), field.dataType())); diff --git a/java/vortex-spark/src/test/java/dev/vortex/spark/VortexOptionsTest.java b/java/vortex-spark/src/test/java/dev/vortex/spark/VortexOptionsTest.java new file mode 100644 index 00000000000..731407d007f --- /dev/null +++ b/java/vortex-spark/src/test/java/dev/vortex/spark/VortexOptionsTest.java @@ -0,0 +1,269 @@ +// SPDX-License-Identifier: Apache-2.0 +// SPDX-FileCopyrightText: Copyright the Vortex contributors + +package dev.vortex.spark; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotEquals; +import static org.junit.jupiter.api.Assertions.assertSame; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.io.ByteArrayInputStream; +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.io.ObjectInputStream; +import java.io.ObjectOutputStream; +import java.util.LinkedHashMap; +import java.util.Map; +import java.util.Optional; +import org.apache.spark.sql.util.CaseInsensitiveStringMap; +import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.api.Test; + +/** + * Unit tests for {@link VortexOptions}. + * + *

Two properties matter most: options resolve case-insensitively, because Spark lower-cases the keys of the map it + * hands to a table; and instances survive Java serialization, because they cross the boundary to the executors inside + * {@code VortexFilePartition} and the reader/writer factories. + */ +final class VortexOptionsTest { + + @Test + @DisplayName("Worker threads default to 4 and are read case-insensitively") + void workerThreadsCaseInsensitive() { + assertEquals(4, VortexOptions.empty().workerThreads()); + assertEquals(16, VortexOptions.of(Map.of("vortex.workerThreads", "16")).workerThreads()); + // The spelling that actually arrives, after Spark has lower-cased the keys. + assertEquals(16, VortexOptions.of(Map.of("vortex.workerthreads", "16")).workerThreads()); + assertEquals(16, VortexOptions.of(Map.of("VORTEX.WORKERTHREADS", "16")).workerThreads()); + } + + @Test + @DisplayName("Worker threads accept zero but reject a negative count") + void workerThreadsRejectNegative() { + assertEquals(0, VortexOptions.of(Map.of("vortex.workerThreads", "0")).workerThreads()); + + IllegalArgumentException thrown = assertThrows( + IllegalArgumentException.class, + () -> VortexOptions.of(Map.of("vortex.workerThreads", "-1")).workerThreads()); + assertTrue(thrown.getMessage().contains("vortex.workerThreads"), thrown.getMessage()); + } + + @Test + @DisplayName("A non-numeric worker thread count is rejected naming the option and the value") + void workerThreadsRejectNonNumeric() { + IllegalArgumentException thrown = assertThrows( + IllegalArgumentException.class, + () -> VortexOptions.of(Map.of("vortex.workerThreads", "eight")).workerThreads()); + + assertTrue(thrown.getMessage().contains("vortex.workerThreads"), thrown.getMessage()); + assertTrue(thrown.getMessage().contains("eight"), thrown.getMessage()); + } + + @Test + @DisplayName("Write batch size defaults to 2048 and honours the legacy key") + void writeBatchSize() { + assertEquals(2048, VortexOptions.empty().writeBatchSize()); + assertEquals( + 4096, + VortexOptions.of(Map.of("vortex.write.batch.size", "4096")).writeBatchSize()); + assertEquals(4096, VortexOptions.of(Map.of("batch.size", "4096")).writeBatchSize()); + assertEquals( + 4096, + VortexOptions.of(Map.of("VORTEX.WRITE.BATCH.SIZE", "4096")).writeBatchSize()); + } + + @Test + @DisplayName("The documented write batch size key wins over the legacy one") + void writeBatchSizePrefersCurrentKey() { + VortexOptions options = VortexOptions.of(Map.of("vortex.write.batch.size", "4096", "batch.size", "1024")); + + assertEquals(4096, options.writeBatchSize()); + } + + @Test + @DisplayName("An out-of-range write batch size falls back to the default and is reported") + void writeBatchSizeOutOfRange() { + VortexOptions tooSmall = VortexOptions.of(Map.of("vortex.write.batch.size", "0")); + VortexOptions tooLarge = VortexOptions.of(Map.of("vortex.write.batch.size", "65537")); + + assertEquals(2048, tooSmall.writeBatchSize()); + assertEquals( + Optional.of(new VortexOptions.RejectedOption("vortex.write.batch.size", 0)), + tooSmall.rejectedWriteBatchSize()); + assertEquals(2048, tooLarge.writeBatchSize()); + assertEquals( + Optional.of(new VortexOptions.RejectedOption("vortex.write.batch.size", 65537)), + tooLarge.rejectedWriteBatchSize()); + + VortexOptions inRange = VortexOptions.of(Map.of("vortex.write.batch.size", "4096")); + assertEquals(Optional.empty(), inRange.rejectedWriteBatchSize()); + + // An out-of-range legacy value is reported under the key the user actually set. + VortexOptions legacy = VortexOptions.of(Map.of("batch.size", "999999")); + assertEquals(2048, legacy.writeBatchSize()); + assertEquals( + Optional.of(new VortexOptions.RejectedOption("batch.size", 999999)), legacy.rejectedWriteBatchSize()); + } + + @Test + @DisplayName("Session provider is empty when unset or blank") + void sessionProvider() { + assertEquals(Optional.empty(), VortexOptions.empty().sessionProvider()); + assertEquals( + Optional.empty(), + VortexOptions.of(Map.of("vortex.session.provider", "")).sessionProvider()); + assertEquals( + Optional.of("com.example.Provider"), + VortexOptions.of(Map.of("vortex.session.provider", "com.example.Provider")) + .sessionProvider()); + assertEquals( + Optional.of("com.example.Provider"), + VortexOptions.of(Map.of("vortex.session.PROVIDER", "com.example.Provider")) + .sessionProvider()); + } + + @Test + @DisplayName("An override replaces the option it means to, whatever the original spelling") + void overrideReplacesDifferentlySpelledKey() { + VortexOptions table = VortexOptions.of(Map.of("vortex.workerThreads", "4")); + + VortexOptions scan = table.withOverrides(Map.of("vortex.workerthreads", "16")); + + assertEquals(16, scan.workerThreads()); + // The stale spelling must be gone, or a case-sensitive reader would still see the old value. + assertEquals(1, scan.asMap().size()); + } + + @Test + @DisplayName("Overrides keep unrelated options and leave the receiver untouched") + void overridesAreAdditiveAndNonMutating() { + Map initial = new LinkedHashMap<>(); + initial.put("aws_region", "us-east-1"); + initial.put("vortex.workerThreads", "4"); + VortexOptions table = VortexOptions.of(initial); + + VortexOptions scan = table.withOverrides(Map.of("vortex.workerThreads", "8")); + + assertEquals("us-east-1", scan.asMap().get("aws_region")); + assertEquals(8, scan.workerThreads()); + assertEquals(4, table.workerThreads()); + } + + @Test + @DisplayName("Overriding with nothing returns the same instance") + void emptyOverridesReturnSameInstance() { + VortexOptions options = VortexOptions.of(Map.of("vortex.workerThreads", "4")); + + assertSame(options, options.withOverrides(Map.of())); + } + + @Test + @DisplayName("Survives Java serialization, including the transient case-insensitive view") + void roundTripsThroughSerialization() throws IOException, ClassNotFoundException { + VortexOptions original = VortexOptions.of(Map.of("vortex.workerThreads", "16", "aws_region", "us-east-1")); + // Resolve first, so the transient case-insensitive view is populated: serializing a fresh + // instance would pass even if that field were not transient. + assertEquals(16, original.workerThreads()); + + VortexOptions restored = roundTrip(original); + + assertEquals(original, restored); + // Resolved after deserialization, so the transient lookup map must have been rebuilt. + assertEquals(16, restored.workerThreads()); + assertEquals("us-east-1", restored.asMap().get("aws_region")); + } + + @Test + @DisplayName("Equality and hashing follow the underlying options") + void equalityFollowsOptions() { + VortexOptions one = VortexOptions.of(Map.of("vortex.workerThreads", "4")); + VortexOptions same = VortexOptions.of(Map.of("vortex.workerThreads", "4")); + VortexOptions other = VortexOptions.of(Map.of("vortex.workerThreads", "8")); + + assertEquals(one, same); + assertEquals(one.hashCode(), same.hashCode()); + assertNotEquals(one, other); + assertEquals(VortexOptions.empty(), VortexOptions.of(Map.of())); + } + + @Test + @DisplayName("The raw map is what the native bindings receive, unchanged") + void asMapPreservesOriginalSpelling() { + VortexOptions options = VortexOptions.of(Map.of("aws_region", "us-east-1")); + + assertEquals(Map.of("aws_region", "us-east-1"), options.asMap()); + } + + @Test + @DisplayName("A valid current key wins even when the legacy key holds garbage") + void currentKeyShortCircuitsLegacyKey() { + VortexOptions options = + VortexOptions.of(Map.of("vortex.write.batch.size", "4096", "batch.size", "not-a-number")); + + assertEquals(4096, options.writeBatchSize()); + assertEquals(Optional.empty(), options.rejectedWriteBatchSize()); + } + + @Test + @DisplayName("Surrounding whitespace in a value is tolerated") + void trimsValues() { + assertEquals( + 16, VortexOptions.of(Map.of("vortex.workerThreads", " 16 ")).workerThreads()); + assertEquals( + 4096, + VortexOptions.of(Map.of("vortex.write.batch.size", " 4096 ")).writeBatchSize()); + } + + @Test + @DisplayName("The map handed to the native bindings cannot be mutated") + void asMapIsImmutable() { + Map exposed = + VortexOptions.of(Map.of("aws_region", "us-east-1")).asMap(); + + assertThrows(UnsupportedOperationException.class, () -> exposed.put("aws_region", "eu-west-1")); + } + + @Test + @DisplayName("Wrapping copies the map, so later changes to the caller's map are not observed") + void ofCopiesTheSuppliedMap() { + Map mutable = new LinkedHashMap<>(); + mutable.put("vortex.workerThreads", "4"); + VortexOptions options = VortexOptions.of(mutable); + + mutable.put("vortex.workerThreads", "16"); + + assertEquals(4, options.workerThreads()); + } + + @Test + @DisplayName("An override supplied through Spark's own map displaces the original spelling") + void overrideThroughSparkMapDisplacesOriginal() { + // This is the shape production sees: Spark hands the table a CaseInsensitiveStringMap, whose + // keys are already lower-cased, to override the table-level options. + VortexOptions table = VortexOptions.of(Map.of("vortex.workerThreads", "4")); + + VortexOptions scan = table.withOverrides(new CaseInsensitiveStringMap(Map.of("vortex.workerThreads", "16"))); + + assertEquals(16, scan.workerThreads()); + assertEquals(1, scan.asMap().size(), scan.asMap().toString()); + } + + @Test + @DisplayName("Options are rejected rather than silently treated as empty") + void ofRejectsNull() { + assertThrows(NullPointerException.class, () -> VortexOptions.of(null)); + } + + private static VortexOptions roundTrip(VortexOptions options) throws IOException, ClassNotFoundException { + ByteArrayOutputStream bytes = new ByteArrayOutputStream(); + try (ObjectOutputStream out = new ObjectOutputStream(bytes)) { + out.writeObject(options); + } + try (ObjectInputStream in = new ObjectInputStream(new ByteArrayInputStream(bytes.toByteArray()))) { + return (VortexOptions) in.readObject(); + } + } +} diff --git a/java/vortex-spark/src/test/java/dev/vortex/spark/VortexTableTest.java b/java/vortex-spark/src/test/java/dev/vortex/spark/VortexTableTest.java index 8c23d6d874b..5f7f24155b9 100644 --- a/java/vortex-spark/src/test/java/dev/vortex/spark/VortexTableTest.java +++ b/java/vortex-spark/src/test/java/dev/vortex/spark/VortexTableTest.java @@ -42,7 +42,7 @@ final class VortexTableTest { }); private static VortexTable tableFor(String... paths) { - return new VortexTable(ImmutableList.copyOf(paths), SCHEMA, Map.of(), new Transform[0]); + return new VortexTable(ImmutableList.copyOf(paths), SCHEMA, VortexOptions.empty(), new Transform[0]); } @Test @@ -66,7 +66,7 @@ void nameJoinsAllPaths() { @DisplayName("Schema and partitioning are returned as supplied") void schemaAndPartitioningRoundTrip() { Transform[] transforms = new Transform[] {Expressions.identity("year")}; - VortexTable table = new VortexTable(ImmutableList.of("/tbl"), SCHEMA, Map.of(), transforms); + VortexTable table = new VortexTable(ImmutableList.of("/tbl"), SCHEMA, VortexOptions.empty(), transforms); assertEquals(SCHEMA, table.schema()); assertArrayEquals(transforms, table.partitioning()); @@ -122,6 +122,27 @@ void writeBuilderRejectsNoPaths() { assertThrows(NoSuchElementException.class, () -> table.newWriteBuilder(writeInfo())); } + @Test + @DisplayName("Scan options override the table's own options, whatever the spelling") + void scanOptionsOverrideTableOptions() { + VortexTable table = new VortexTable( + ImmutableList.of("/data/a.vortex"), + SCHEMA, + VortexOptions.of(Map.of(VortexOptions.WORKER_THREADS, "4", "aws_region", "us-east-1")), + new Transform[0]); + + var builder = table.newScanBuilder(new CaseInsensitiveStringMap(Map.of(VortexOptions.WORKER_THREADS, "16"))); + VortexFilePartition partition = + (VortexFilePartition) ((org.apache.spark.sql.connector.read.SupportsReportStatistics) builder.build()) + .toBatch() + .planInputPartitions()[0]; + + assertEquals(16, partition.formatOptions().workerThreads()); + // The unrelated table option survives, and the overridden one is not left behind twice. + assertEquals("us-east-1", partition.formatOptions().asMap().get("aws_region")); + assertEquals(2, partition.formatOptions().asMap().size()); + } + private static LogicalWriteInfo writeInfo() { return new LogicalWriteInfo() { @Override diff --git a/java/vortex-spark/src/test/java/dev/vortex/spark/read/VortexBatchExecTest.java b/java/vortex-spark/src/test/java/dev/vortex/spark/read/VortexBatchExecTest.java index 967940567b2..0abda013b49 100644 --- a/java/vortex-spark/src/test/java/dev/vortex/spark/read/VortexBatchExecTest.java +++ b/java/vortex-spark/src/test/java/dev/vortex/spark/read/VortexBatchExecTest.java @@ -7,6 +7,7 @@ import static org.junit.jupiter.api.Assertions.assertInstanceOf; import dev.vortex.spark.VortexFilePartition; +import dev.vortex.spark.VortexOptions; import java.util.List; import java.util.Map; import org.apache.spark.sql.connector.catalog.Column; @@ -29,7 +30,7 @@ final class VortexBatchExecTest { Column.create("name", org.apache.spark.sql.types.DataTypes.StringType)); private static VortexBatchExec execFor(List paths) { - return new VortexBatchExec(paths, COLUMNS, Map.of(), new Predicate[0]); + return new VortexBatchExec(paths, COLUMNS, VortexOptions.empty(), new Predicate[0]); } @Test @@ -85,11 +86,14 @@ void partitionValuesParsedPerFile() { @Test @DisplayName("Format options are propagated to every partition") void formatOptionsPropagated() { - Map options = Map.of("vortex.workerThreads", "8"); + VortexOptions options = VortexOptions.of(Map.of(VortexOptions.WORKER_THREADS, "8")); VortexBatchExec exec = new VortexBatchExec(List.of("/data/a.vortex"), COLUMNS, options, new Predicate[0]); VortexFilePartition partition = (VortexFilePartition) exec.planInputPartitions()[0]; - assertEquals("8", partition.formatOptions().get("vortex.workerThreads")); + assertEquals(8, partition.formatOptions().workerThreads()); + // Also pin the raw spelling: the native bindings receive asMap(), so a key mangled in + // transit would still resolve through the case-insensitive accessor above. + assertEquals("8", partition.formatOptions().asMap().get(VortexOptions.WORKER_THREADS)); } }