From 210dac6c44eca1b4ed4135b431fd76f1a9713d6a Mon Sep 17 00:00:00 2001 From: jackylee Date: Wed, 5 Aug 2026 20:19:30 +0800 Subject: [PATCH] refactor(spark): resolve format options through a VortexOptions type Spark lower-cases the keys of the CaseInsensitiveStringMap it hands to a table, so an option a user spelled vortex.workerThreads arrived as vortex.workerthreads. Every call site that read an option had to know that, and none did: the worker-thread count was looked up case-sensitively and always fell back to its default, and VortexTable merged scan options over table options with a plain putAll, leaving the same option present twice under two spellings. Introduce VortexOptions, following the shape of Spark's own ParquetOptions: it is Serializable, holds its CaseInsensitiveStringMap as transient and rebuilds it on demand, and gives every option one place that names it, documents its default and validates it. withOverrides matches keys case-insensitively so an override replaces the option it means to. Thread it through the read and write paths in place of the raw map, converting back with asMap() only where the native bindings are called, and drop the camelCase seeding VortexScanBuilder used to compensate. VortexSparkSession.get and the writer's batch-size resolution now go through it too, so vortex.session.provider and vortex.write.batch.size resolve case-insensitively as well. Signed-off-by: jackylee --- .../dev/vortex/spark/VortexDataSourceV2.java | 12 +- .../dev/vortex/spark/VortexFilePartition.java | 5 +- .../java/dev/vortex/spark/VortexOptions.java | 203 +++++++++++++ .../dev/vortex/spark/VortexSparkSession.java | 17 +- .../java/dev/vortex/spark/VortexTable.java | 11 +- .../vortex/spark/read/VortexBatchExec.java | 15 +- .../spark/read/VortexPartitionReader.java | 5 +- .../read/VortexPartitionReaderFactory.java | 12 +- .../dev/vortex/spark/read/VortexScan.java | 7 +- .../vortex/spark/read/VortexScanBuilder.java | 13 +- .../write/PartitionedVortexDataWriter.java | 6 +- .../vortex/spark/write/VortexBatchWrite.java | 12 +- .../vortex/spark/write/VortexDataWriter.java | 47 ++- .../spark/write/VortexDataWriterFactory.java | 15 +- .../spark/write/VortexWriteBuilder.java | 6 +- .../spark/VortexDataSourceStatsTest.java | 3 +- .../dev/vortex/spark/VortexOptionsTest.java | 269 ++++++++++++++++++ .../dev/vortex/spark/VortexTableTest.java | 25 +- .../spark/read/VortexBatchExecTest.java | 10 +- 19 files changed, 587 insertions(+), 106 deletions(-) create mode 100644 java/vortex-spark/src/main/java/dev/vortex/spark/VortexOptions.java create mode 100644 java/vortex-spark/src/test/java/dev/vortex/spark/VortexOptionsTest.java 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)); } }