This is an automated email from the ASF dual-hosted git repository.
voonhous pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git
The following commit(s) were added to refs/heads/master by this push:
new 0fc240bf5f95 test: add Vortex reader/writer round-trip test (#19182)
0fc240bf5f95 is described below
commit 0fc240bf5f95d2fa979e3c8e01f66a65724abfe1
Author: Y Ethan Guo <[email protected]>
AuthorDate: Wed Jul 8 06:13:58 2026 -0700
test: add Vortex reader/writer round-trip test (#19182)
* feat: Add VORTEX file format enum, storage configs, and Maven dependency
management
Closes #18624
* feat: Wire VORTEX into reader/writer factory dispatch and IOFactory
Closes #18625
* feat: Add Vortex base writer and Spark InternalRow writer
Closes #18626
* feat: Add Vortex Spark reader, record iterator, and resource closer
Closes #18627
* feat: Add Vortex Hive InputFormat stubs and wire Hive dispatch
Mirror the Lance Hive integration for the VORTEX base file format:
- Add HoodieVortexInputFormat (COW) and HoodieVortexRealtimeInputFormat
(MOR) as catalog/metastore registration stubs; reading through the Hive
InputFormat is not yet supported and throws, directing users to the Spark
datasource path.
- Wire VORTEX into HoodieInputFormatUtils dispatch: getInputFormat (by
format and by extension), getInputFormatClassName,
getOutputFormatClassName
and getSerDeClassName (reusing the Parquet OutputFormat/SerDe like Lance).
- Guard the Hive reader context so .vortex files fail fast with
VORTEX_SPARK_ONLY_ERROR_MSG instead of being mis-read.
VortexUtils (key iteration, schema reads, row counts, filterRowKeys) is
already implemented at parity with LanceUtils.
* feat: Add Vortex Spark datasource reader and wire into SparkAdapter
Add the query-time Vortex reader for the Spark datasource path, mirroring
the
Lance integration for the initial Vortex scope (standard data types, no
filter
pushdown/data skipping, no BLOB/VECTOR handling).
- Create SparkVortexReaderBase (hudi-spark-common) implementing
SparkColumnarFileReader: opens the file via the dev.vortex.api Session,
projects
the requested columns, converts Arrow batches to rows through the existing
VortexRecordIterator, and applies null-padding/cast projections plus
partition
column appending for schema evolution. Handles the partition-only /
all-new-column
case by emitting empty rows sized from DataSource.numRows().
- Add SparkAdapter.createVortexFileReader; override in the Spark
3.5/4.0/4.1/4.2
adapters (Some) and 3.3/3.4 (None, connector artifact unavailable),
matching the
vortex-spark test matrix in the pom.
- Extend MultipleColumnarFileFormatReader with a fourth vortexReader and a
VORTEX
dispatch case; thread it through HoodieFileGroupReaderBasedFileFormat,
HoodieMergeOnReadRDDV2, and SparkReaderContextFactory (plus the
single-format
VORTEX branches) and update the file-group reader test constructor.
* test: add Vortex writer and reader round-trip test
Adds a file-level round trip through HoodieSparkVortexWriter and
HoodieSparkVortexReader against the real dev.vortex.api: a multi-batch
write+read of long/string/double columns and a projected (column-subset)
read. Gated by -Dvortex.skip.tests.
closes #18630
* test: strengthen Vortex round-trip test per review
- Wrap HoodieStorage in try-with-resources so it closes even when an
assertion
throws (previously closed only on the happy path).
- Project the non-leading "name" column in testProjectedRead so the
assertion
actually pins projection; a leading-column ("id") projection read back at
index 0 whether or not the reader narrowed to the requested schema.
---------
Co-authored-by: voon <[email protected]>
---
hudi-client/hudi-spark-client/pom.xml | 44 +++
.../hudi/io/storage/HoodieSparkVortexReader.java | 205 ++++++++++++
.../hudi/io/storage/HoodieSparkVortexWriter.java | 259 ++++++++++++++++
.../hudi/io/storage/VortexRecordIterator.java | 199 ++++++++++++
.../hudi/io/storage/VortexResourceCloser.java | 57 ++++
.../client/common/SparkReaderContextFactory.java | 6 +-
.../io/storage/HoodieSparkFileReaderFactory.java | 11 +
.../io/storage/HoodieSparkFileWriterFactory.java | 22 ++
.../hudi/MultipleColumnarFileFormatReader.scala | 11 +-
.../org/apache/spark/sql/hudi/SparkAdapter.scala | 14 +
.../hudi/common/config/HoodieStorageConfig.java | 42 +++
.../apache/hudi/common/model/HoodieFileFormat.java | 10 +-
.../org/apache/hudi/common/util/VortexUtils.java | 203 ++++++++++++
.../core/io/storage/HoodieFileReaderFactory.java | 10 +
.../core/io/storage/HoodieFileWriterFactory.java | 10 +
.../hudi/core/io/storage/HoodieIOFactory.java | 2 +
hudi-hadoop-common/pom.xml | 44 +++
.../hudi/io/vortex/HoodieBaseVortexWriter.java | 345 +++++++++++++++++++++
.../io/storage/hadoop/HoodieHadoopIOFactory.java | 3 +
.../hudi/hadoop/HiveHoodieReaderContext.java | 3 +
.../hudi/hadoop/HoodieVortexInputFormat.java | 67 ++++
.../realtime/HoodieVortexRealtimeInputFormat.java | 64 ++++
.../hudi/hadoop/utils/HoodieInputFormatUtils.java | 23 ++
hudi-spark-datasource/hudi-spark-common/pom.xml | 44 +++
.../datasources/vortex/SparkVortexReaderBase.scala | 203 ++++++++++++
.../org/apache/hudi/HoodieMergeOnReadRDDV2.scala | 5 +-
.../HoodieFileGroupReaderBasedFileFormat.scala | 5 +-
hudi-spark-datasource/hudi-spark/pom.xml | 43 +++
.../storage/TestVortexReaderWriterRoundTrip.java | 141 +++++++++
.../read/TestHoodieFileGroupReaderOnSpark.scala | 3 +-
.../apache/spark/sql/adapter/Spark3_3Adapter.scala | 7 +
.../apache/spark/sql/adapter/Spark3_4Adapter.scala | 8 +
.../apache/spark/sql/adapter/Spark3_5Adapter.scala | 9 +
.../apache/spark/sql/adapter/Spark4_0Adapter.scala | 8 +
.../apache/spark/sql/adapter/Spark4_1Adapter.scala | 8 +
.../apache/spark/sql/adapter/Spark4_2Adapter.scala | 8 +
pom.xml | 36 +++
37 files changed, 2174 insertions(+), 8 deletions(-)
diff --git a/hudi-client/hudi-spark-client/pom.xml
b/hudi-client/hudi-spark-client/pom.xml
index a54e33c9857a..21697a660957 100644
--- a/hudi-client/hudi-spark-client/pom.xml
+++ b/hudi-client/hudi-spark-client/pom.xml
@@ -300,4 +300,48 @@
</resource>
</resources>
</build>
+
+ <profiles>
+ <!-- Vortex requires Java 17 (vortex-jni/vortex-spark ship Java 17
bytecode). Compile the Vortex
+ reader/writer sources under src/main/java-vortex and pull the
vortex-spark connector only on
+ JDK 17; Java 11 builds skip them. The factory dispatch
(HoodieSparkFile{Reader,Writer}Factory)
+ instantiates these classes reflectively so it stays Java 11
compilable. -->
+ <profile>
+ <id>vortex</id>
+ <activation>
+ <jdk>[17,)</jdk>
+ </activation>
+ <dependencies>
+ <dependency>
+ <groupId>dev.vortex</groupId>
+ <artifactId>${vortex.spark.artifact}</artifactId>
+ </dependency>
+ </dependencies>
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.codehaus.mojo</groupId>
+ <artifactId>build-helper-maven-plugin</artifactId>
+ <executions>
+ <execution>
+ <id>add-vortex-source</id>
+ <phase>generate-sources</phase>
+ <goals>
+ <goal>add-source</goal>
+ </goals>
+ <configuration>
+ <!-- Vortex is only supported on Spark 3.5+/4.x; skip the
source on Spark 3.3/3.4
+ (vortex.skip.tests=true) where its Spark 3.5+ APIs are
unavailable. -->
+ <skipAddSource>${vortex.skip.tests}</skipAddSource>
+ <sources>
+ <source>src/main/java-vortex</source>
+ </sources>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
+ </plugins>
+ </build>
+ </profile>
+ </profiles>
</project>
diff --git
a/hudi-client/hudi-spark-client/src/main/java-vortex/org/apache/hudi/io/storage/HoodieSparkVortexReader.java
b/hudi-client/hudi-spark-client/src/main/java-vortex/org/apache/hudi/io/storage/HoodieSparkVortexReader.java
new file mode 100644
index 000000000000..d5cb85fd9bb8
--- /dev/null
+++
b/hudi-client/hudi-spark-client/src/main/java-vortex/org/apache/hudi/io/storage/HoodieSparkVortexReader.java
@@ -0,0 +1,205 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.io.storage;
+
+import org.apache.hudi.HoodieSchemaConversionUtils;
+import org.apache.hudi.common.bloom.BloomFilter;
+import org.apache.hudi.common.config.HoodieStorageConfig;
+import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.common.model.HoodieSparkRecord;
+import org.apache.hudi.common.schema.HoodieSchema;
+import org.apache.hudi.common.schema.HoodieSchemaUtils;
+import org.apache.hudi.common.util.collection.ClosableIterator;
+import org.apache.hudi.common.util.collection.CloseableMappingIterator;
+import org.apache.hudi.common.util.collection.Pair;
+import org.apache.hudi.exception.HoodieException;
+import org.apache.hudi.exception.HoodieIOException;
+import org.apache.hudi.io.memory.HoodieArrowAllocator;
+import org.apache.hudi.storage.StoragePath;
+
+import dev.vortex.api.DataSource;
+import dev.vortex.api.Scan;
+import dev.vortex.api.ScanOptions;
+import dev.vortex.api.Session;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.arrow.memory.BufferAllocator;
+import org.apache.arrow.vector.types.pojo.Schema;
+import org.apache.spark.sql.catalyst.InternalRow;
+import org.apache.spark.sql.catalyst.expressions.UnsafeRow;
+import org.apache.spark.sql.types.StructType;
+import org.apache.spark.sql.util.ArrowUtils$;
+
+import java.io.IOException;
+import java.net.URI;
+import java.util.HashSet;
+import java.util.Set;
+
+import static org.apache.hudi.common.util.TypeUtils.unsafeCast;
+
+/**
+ * {@link HoodieSparkFileReader} implementation for Vortex file format, backed
by the vortex-jni
+ * {@code dev.vortex.api} reader (Session / DataSource / Scan).
+ *
+ * <p>NOTE: vortex-jni 0.76.0 exposes no file-level custom-metadata API, so
the Hudi bloom filter
+ * and min/max record-key footers are not available for Vortex files. {@link
#readBloomFilter()}
+ * returns {@code null} and {@link #readMinMaxRecordKeys()} is unsupported.
+ * TODO(#18623): surface bloom/min-max footers once Vortex supports file
metadata.
+ */
+@Slf4j
+public class HoodieSparkVortexReader implements HoodieSparkFileReader {
+ private final StoragePath path;
+ private final String uri;
+ private final long dataAllocatorSize;
+ // Session/DataSource are held so their native resources are not GC'd while
this reader is open;
+ // neither is AutoCloseable in vortex-jni 0.76.0 (cleanup is implicit).
+ private final Session metadataSession;
+ private final DataSource metadataDataSource;
+ private final BufferAllocator metadataAllocator;
+
+ public HoodieSparkVortexReader(StoragePath path) {
+ this(path,
Long.parseLong(HoodieStorageConfig.VORTEX_READ_ALLOCATOR_SIZE_BYTES.defaultValue()));
+ }
+
+ public HoodieSparkVortexReader(StoragePath path, long dataAllocatorSize) {
+ this.path = path;
+ this.uri = toVortexUri(path);
+ this.dataAllocatorSize = dataAllocatorSize;
+ this.metadataAllocator = HoodieArrowAllocator.newChildAllocator(
+ getClass().getSimpleName() + "-meta-" + path.getName(),
+
Long.parseLong(HoodieStorageConfig.VORTEX_READ_METADATA_ALLOCATOR_SIZE_BYTES.defaultValue()));
+ try {
+ this.metadataSession = Session.create();
+ this.metadataDataSource = DataSource.open(metadataSession, uri);
+ } catch (Exception e) {
+ metadataAllocator.close();
+ throw new HoodieException("Failed to open Vortex file: " + path, e);
+ }
+ }
+
+ @Override
+ public String[] readMinMaxRecordKeys() {
+ throw new UnsupportedOperationException(
+ "Min/max record keys are not available for Vortex files (no
file-metadata support in vortex-jni 0.76.0)");
+ }
+
+ @Override
+ public BloomFilter readBloomFilter() {
+ // Vortex files carry no Hudi bloom filter footer yet; callers treat null
as "no bloom filter".
+ return null;
+ }
+
+ @Override
+ public Set<Pair<String, Long>> filterRowKeys(Set<String> candidateRowKeys) {
+ Set<Pair<String, Long>> result = new HashSet<>();
+ long position = 0;
+
+ try (ClosableIterator<String> keyIterator = getRecordKeyIterator()) {
+ while (keyIterator.hasNext()) {
+ String recordKey = keyIterator.next();
+ if (candidateRowKeys == null || candidateRowKeys.isEmpty()
+ || candidateRowKeys.contains(recordKey)) {
+ result.add(Pair.of(recordKey, position));
+ }
+ position++;
+ }
+ } catch (IOException e) {
+ throw new HoodieIOException("Failed to filter row keys from Vortex file:
" + path, e);
+ }
+
+ return result;
+ }
+
+ @Override
+ public ClosableIterator<HoodieRecord<InternalRow>>
getRecordIterator(HoodieSchema readerSchema, HoodieSchema requestedSchema)
throws IOException {
+ return getRecordIterator(requestedSchema);
+ }
+
+ @Override
+ public ClosableIterator<HoodieRecord<InternalRow>>
getRecordIterator(HoodieSchema schema) throws IOException {
+ ClosableIterator<UnsafeRow> iterator = getUnsafeRowIterator(schema);
+ return new CloseableMappingIterator<>(iterator, data -> unsafeCast(new
HoodieSparkRecord(data)));
+ }
+
+ @Override
+ public ClosableIterator<String> getRecordKeyIterator() throws IOException {
+ HoodieSchema recordKeySchema = HoodieSchemaUtils.getRecordKeySchema();
+ ClosableIterator<UnsafeRow> iterator =
getUnsafeRowIterator(recordKeySchema);
+ return new CloseableMappingIterator<>(iterator, data ->
data.getUTF8String(0).toString());
+ }
+
+ private ClosableIterator<UnsafeRow> getUnsafeRowIterator(HoodieSchema
requestedSchema) {
+ StructType requestedSparkSchema =
HoodieSchemaConversionUtils.convertHoodieSchemaToStructType(requestedSchema);
+
+ BufferAllocator allocator = HoodieArrowAllocator.newChildAllocator(
+ getClass().getSimpleName() + "-data-" + path.getName(),
dataAllocatorSize);
+
+ try {
+ Session session = Session.create();
+ DataSource dataSource = DataSource.open(session, uri);
+ // Read all columns; VortexRecordIterator projects the requested columns
by name.
+ Scan scan = dataSource.scan(ScanOptions.of());
+ return new VortexRecordIterator(allocator, session, dataSource, scan,
requestedSparkSchema, uri);
+ } catch (Exception e) {
+ allocator.close();
+ throw new HoodieException("Failed to create Vortex reader for: " + path,
e);
+ }
+ }
+
+ @Override
+ public HoodieSchema getSchema() {
+ try {
+ Schema arrowSchema = metadataDataSource.arrowSchema(metadataAllocator);
+ StructType structType = (StructType)
ArrowUtils$.MODULE$.fromArrowSchema(arrowSchema);
+ return
HoodieSchemaConversionUtils.convertStructTypeToHoodieSchema(structType,
"record", "", false);
+ } catch (Exception e) {
+ throw new HoodieException("Failed to read schema from Vortex file: " +
path, e);
+ }
+ }
+
+ @Override
+ public void close() {
+ try {
+ metadataAllocator.close();
+ } catch (Exception e) {
+ log.warn("Error while closing Vortex metadata allocator: {}",
e.getMessage());
+ }
+ }
+
+ @Override
+ public long getTotalRecords() {
+ try {
+ return metadataDataSource.rowCount().asOptional().orElseThrow(
+ () -> new HoodieException("Vortex row count unavailable for file: "
+ path));
+ } catch (Exception e) {
+ throw new HoodieException("Failed to get row count from Vortex file: " +
path, e);
+ }
+ }
+
+ /**
+ * Vortex requires a well-formed URL. Remote-store {@link StoragePath}s
already carry a scheme
+ * (s3://, hdfs://, ...); local paths may not, so qualify them as {@code
file://} URIs.
+ */
+ private static String toVortexUri(StoragePath path) {
+ URI parsed = path.toUri();
+ if (parsed.getScheme() == null) {
+ return new java.io.File(parsed.getPath()).toURI().toString();
+ }
+ return parsed.toString();
+ }
+}
diff --git
a/hudi-client/hudi-spark-client/src/main/java-vortex/org/apache/hudi/io/storage/HoodieSparkVortexWriter.java
b/hudi-client/hudi-spark-client/src/main/java-vortex/org/apache/hudi/io/storage/HoodieSparkVortexWriter.java
new file mode 100644
index 000000000000..4467c30d2496
--- /dev/null
+++
b/hudi-client/hudi-spark-client/src/main/java-vortex/org/apache/hudi/io/storage/HoodieSparkVortexWriter.java
@@ -0,0 +1,259 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.io.storage;
+
+import org.apache.hudi.common.bloom.BloomFilter;
+import org.apache.hudi.common.config.HoodieStorageConfig;
+import org.apache.hudi.common.engine.TaskContextSupplier;
+import org.apache.hudi.common.model.HoodieKey;
+import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.io.storage.row.HoodieBloomFilterRowWriteSupport;
+import org.apache.hudi.io.storage.row.HoodieInternalRowFileWriter;
+import org.apache.hudi.io.vortex.HoodieBaseVortexWriter;
+import org.apache.hudi.storage.HoodieStorage;
+import org.apache.hudi.storage.StoragePath;
+
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import org.apache.arrow.vector.VectorSchemaRoot;
+import org.apache.arrow.vector.types.pojo.Schema;
+import org.apache.spark.sql.catalyst.InternalRow;
+import org.apache.spark.sql.execution.arrow.ArrowWriter$;
+import org.apache.spark.sql.types.StructType;
+import org.apache.spark.sql.util.ArrowUtils$;
+import org.apache.spark.unsafe.types.UTF8String;
+
+import java.io.IOException;
+import java.util.function.Function;
+
+import static
org.apache.hudi.common.model.HoodieRecord.HoodieMetadataField.COMMIT_SEQNO_METADATA_FIELD;
+import static
org.apache.hudi.common.model.HoodieRecord.HoodieMetadataField.COMMIT_TIME_METADATA_FIELD;
+import static
org.apache.hudi.common.model.HoodieRecord.HoodieMetadataField.FILENAME_METADATA_FIELD;
+import static
org.apache.hudi.common.model.HoodieRecord.HoodieMetadataField.PARTITION_PATH_METADATA_FIELD;
+import static
org.apache.hudi.common.model.HoodieRecord.HoodieMetadataField.RECORD_KEY_METADATA_FIELD;
+import static org.apache.hudi.common.util.ValidationUtils.checkArgument;
+
+/**
+ * Spark Vortex file writer implementing {@link HoodieSparkFileWriter} and
{@link HoodieInternalRowFileWriter}.
+ *
+ * This writer integrates with Hudi's storage I/O layer and supports:
+ * - Hudi metadata field population
+ * - Record key tracking (for bloom filters)
+ * - Sequence ID generation
+ * - Min/max record key tracking
+ */
+public class HoodieSparkVortexWriter extends
HoodieBaseVortexWriter<InternalRow, UTF8String>
+ implements HoodieSparkFileWriter, HoodieInternalRowFileWriter {
+
+ private static final String DEFAULT_TIMEZONE = "UTC";
+ private static final long MIN_RECORDS_FOR_SIZE_CHECK = 100L;
+ private static final long MAX_RECORDS_FOR_SIZE_CHECK = 10000L;
+
+ private final StructType sparkSchema;
+ private final Schema arrowSchema;
+ private final UTF8String fileName;
+ private final UTF8String instantTime;
+ private final boolean populateMetaFields;
+ private final Function<Long, String> seqIdGenerator;
+ private final long maxFileSize;
+ private long recordCountForNextSizeCheck = MIN_RECORDS_FOR_SIZE_CHECK;
+
+ @Builder(builderMethodName = "builder")
+ private static HoodieSparkVortexWriter create(
+ StoragePath file,
+ StructType sparkSchema,
+ String instantTime,
+ TaskContextSupplier taskContextSupplier,
+ HoodieStorage storage,
+ boolean populateMetaFields,
+ Option<BloomFilter> bloomFilterOpt,
+ long maxFileSize,
+ long allocatorSize,
+ long flushByteWatermark) {
+ checkArgument(maxFileSize > 0, "maxFileSize must be a positive number");
+ checkArgument(allocatorSize > 0, "allocatorSize must be a positive
number");
+ checkArgument(flushByteWatermark > 0, "flushByteWatermark must be a
positive number");
+ checkArgument(flushByteWatermark < allocatorSize,
+ "flushByteWatermark (" + flushByteWatermark + ") must be less than
allocatorSize ("
+ + allocatorSize + ") so the byte-aware flush prevents
reallocations from exceeding the cap");
+ return new HoodieSparkVortexWriter(file, sparkSchema, instantTime,
+ taskContextSupplier, storage, populateMetaFields, bloomFilterOpt,
maxFileSize,
+ allocatorSize, flushByteWatermark);
+ }
+
+ /**
+ * Public reflection entry point for {@code HoodieSparkFileWriterFactory}.
The factory lives in the
+ * default (Java 11) source set and cannot reference this Vortex-only class
directly, so it invokes
+ * this method reflectively (see the vortex profile that compiles this
source under JDK 17).
+ */
+ public static HoodieSparkVortexWriter newWriter(
+ StoragePath file,
+ StructType sparkSchema,
+ String instantTime,
+ TaskContextSupplier taskContextSupplier,
+ HoodieStorage storage,
+ boolean populateMetaFields,
+ Option<BloomFilter> bloomFilterOpt,
+ long maxFileSize,
+ long allocatorSize,
+ long flushByteWatermark) {
+ return builder()
+ .file(file)
+ .sparkSchema(sparkSchema)
+ .instantTime(instantTime)
+ .taskContextSupplier(taskContextSupplier)
+ .storage(storage)
+ .populateMetaFields(populateMetaFields)
+ .bloomFilterOpt(bloomFilterOpt)
+ .maxFileSize(maxFileSize)
+ .allocatorSize(allocatorSize)
+ .flushByteWatermark(flushByteWatermark)
+ .build();
+ }
+
+ /**
+ * Manually declared builder class to provide default values for optional
parameters.
+ * Lombok fills in the remaining builder methods.
+ */
+ public static class HoodieSparkVortexWriterBuilder {
+ private Option<BloomFilter> bloomFilterOpt = Option.empty();
+ private long maxFileSize =
Long.parseLong(HoodieStorageConfig.VORTEX_MAX_FILE_SIZE.defaultValue());
+ private long allocatorSize =
Long.parseLong(HoodieStorageConfig.VORTEX_WRITE_ALLOCATOR_SIZE_BYTES.defaultValue());
+ private long flushByteWatermark =
Long.parseLong(HoodieStorageConfig.VORTEX_WRITE_FLUSH_BYTE_WATERMARK.defaultValue());
+ }
+
+ private HoodieSparkVortexWriter(StoragePath file,
+ StructType sparkSchema,
+ String instantTime,
+ TaskContextSupplier taskContextSupplier,
+ HoodieStorage storage,
+ boolean populateMetaFields,
+ Option<BloomFilter> bloomFilterOpt,
+ long maxFileSize,
+ long allocatorSize,
+ long flushByteWatermark) {
+ super(file, DEFAULT_BATCH_SIZE, allocatorSize, flushByteWatermark,
+ bloomFilterOpt.map(HoodieBloomFilterRowWriteSupport::new));
+ this.sparkSchema = sparkSchema;
+ this.arrowSchema = ArrowUtils$.MODULE$.toArrowSchema(sparkSchema,
DEFAULT_TIMEZONE, false, false);
+ this.fileName = UTF8String.fromString(file.getName());
+ this.instantTime = UTF8String.fromString(instantTime);
+ this.populateMetaFields = populateMetaFields;
+ this.maxFileSize = maxFileSize;
+ this.seqIdGenerator = recordIndex -> {
+ Integer partitionId = taskContextSupplier.getPartitionIdSupplier().get();
+ return HoodieRecord.generateSequenceId(instantTime, partitionId,
recordIndex);
+ };
+ }
+
+ @Override
+ public void writeRowWithMetadata(HoodieKey key, InternalRow row) throws
IOException {
+ UTF8String recordKey = UTF8String.fromString(key.getRecordKey());
+ bloomFilterWriteSupportOpt.ifPresent(bloomFilterWriteSupport ->
bloomFilterWriteSupport.addKey(recordKey));
+ if (populateMetaFields) {
+ updateRecordMetadata(row, recordKey, key.getPartitionPath(),
getWrittenRecordCount());
+ }
+ super.write(row);
+ }
+
+ @Override
+ public void writeRow(String recordKey, InternalRow row) throws IOException {
+ bloomFilterWriteSupportOpt.ifPresent(bloomFilterWriteSupport ->
+ bloomFilterWriteSupport.addKey(UTF8String.fromString(recordKey)));
+ super.write(row);
+ }
+
+ @Override
+ public void writeRow(UTF8String key, InternalRow row) throws IOException {
+ bloomFilterWriteSupportOpt.ifPresent(bloomFilterWriteSupport ->
bloomFilterWriteSupport.addKey(key));
+ super.write(row);
+ }
+
+ @Override
+ public void writeRow(InternalRow row) throws IOException {
+ super.write(row);
+ }
+
+ @Override
+ protected HoodieBaseVortexWriter.ArrowWriter<InternalRow>
createArrowWriter(VectorSchemaRoot root) {
+ return SparkArrowWriterAdapter.of(ArrowWriter$.MODULE$.create(root));
+ }
+
+ /**
+ * Check if writer can accept more records based on estimated data size.
+ *
+ * @return true if writer can accept more records, false if file size limit
is reached
+ */
+ public boolean canWrite() {
+ long writtenCount = getWrittenRecordCount();
+ if (writtenCount >= recordCountForNextSizeCheck) {
+ long dataSize = getDataSize();
+ long avgRecordSize = Math.max(dataSize / writtenCount, 1);
+ if (dataSize > (maxFileSize - avgRecordSize * 2)) {
+ return false;
+ }
+ recordCountForNextSizeCheck = writtenCount + Math.min(
+ Math.max(MIN_RECORDS_FOR_SIZE_CHECK, (maxFileSize / avgRecordSize -
writtenCount) / 2),
+ MAX_RECORDS_FOR_SIZE_CHECK);
+ }
+ return true;
+ }
+
+ @Override
+ protected Schema getArrowSchema() {
+ return arrowSchema;
+ }
+
+ /**
+ * Update Hudi metadata fields in the InternalRow.
+ */
+ protected void updateRecordMetadata(InternalRow row,
+ UTF8String recordKey,
+ String partitionPath,
+ long recordCount) {
+ row.update(COMMIT_TIME_METADATA_FIELD.ordinal(), instantTime);
+ row.update(COMMIT_SEQNO_METADATA_FIELD.ordinal(),
UTF8String.fromString(seqIdGenerator.apply(recordCount)));
+ row.update(RECORD_KEY_METADATA_FIELD.ordinal(), recordKey);
+ row.update(PARTITION_PATH_METADATA_FIELD.ordinal(),
UTF8String.fromString(partitionPath));
+ row.update(FILENAME_METADATA_FIELD.ordinal(), fileName);
+ }
+
+ @AllArgsConstructor(staticName = "of")
+ private static class SparkArrowWriterAdapter implements
HoodieBaseVortexWriter.ArrowWriter<InternalRow> {
+ // Fully qualified: the inherited HoodieBaseVortexWriter.ArrowWriter
nested type would otherwise
+ // shadow this Spark ArrowWriter import.
+ private final org.apache.spark.sql.execution.arrow.ArrowWriter
sparkArrowWriter;
+
+ @Override
+ public void write(InternalRow row) {
+ sparkArrowWriter.write(row);
+ }
+
+ @Override
+ public void reset() {
+ sparkArrowWriter.reset();
+ }
+
+ @Override
+ public void finishBatch() {
+ sparkArrowWriter.finish();
+ }
+ }
+}
diff --git
a/hudi-client/hudi-spark-client/src/main/java-vortex/org/apache/hudi/io/storage/VortexRecordIterator.java
b/hudi-client/hudi-spark-client/src/main/java-vortex/org/apache/hudi/io/storage/VortexRecordIterator.java
new file mode 100644
index 000000000000..85724f05a936
--- /dev/null
+++
b/hudi-client/hudi-spark-client/src/main/java-vortex/org/apache/hudi/io/storage/VortexRecordIterator.java
@@ -0,0 +1,199 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.io.storage;
+
+import org.apache.hudi.common.util.collection.ClosableIterator;
+import org.apache.hudi.exception.HoodieException;
+
+import dev.vortex.api.DataSource;
+import dev.vortex.api.Partition;
+import dev.vortex.api.Scan;
+import dev.vortex.api.Session;
+import org.apache.arrow.memory.BufferAllocator;
+import org.apache.arrow.vector.FieldVector;
+import org.apache.arrow.vector.VectorSchemaRoot;
+import org.apache.arrow.vector.ipc.ArrowReader;
+import org.apache.spark.sql.catalyst.InternalRow;
+import org.apache.spark.sql.catalyst.expressions.UnsafeProjection;
+import org.apache.spark.sql.catalyst.expressions.UnsafeRow;
+import org.apache.spark.sql.types.StructField;
+import org.apache.spark.sql.types.StructType;
+import org.apache.spark.sql.vectorized.ArrowColumnVector;
+import org.apache.spark.sql.vectorized.ColumnVector;
+import org.apache.spark.sql.vectorized.ColumnarBatch;
+
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Iterator for reading Vortex files (vortex-jni 0.76.0) and converting Arrow
batches to Spark
+ * {@link UnsafeRow}s.
+ *
+ * <p>A {@link Scan} yields a sequence of {@link Partition}s; each partition
is read as an Arrow
+ * {@link ArrowReader} via {@link Partition#scanArrow}, whose {@link
VectorSchemaRoot} is reused
+ * across {@code loadNextBatch} calls. Only the columns present in the
requested Spark schema are
+ * projected (by name) out of each batch, so the scan may read more columns
than requested.
+ *
+ * <p>Lifecycle managed here: the {@link BufferAllocator} and the current
{@link ArrowReader}. The
+ * {@link Session}/{@link DataSource}/{@link Scan} handles are held only to
keep their native
+ * resources alive during iteration (they are not AutoCloseable in vortex-jni
0.76.0).
+ */
+public final class VortexRecordIterator implements ClosableIterator<UnsafeRow>
{
+ private final BufferAllocator allocator;
+ private final Session session;
+ private final DataSource dataSource;
+ private final Scan scan;
+ private final StructType sparkSchema;
+ private final UnsafeProjection projection;
+ private final String path;
+
+ private ArrowReader currentReader;
+ private ColumnVector[] columnVectors;
+ private ColumnarBatch currentBatch;
+ private Iterator<InternalRow> rowIterator;
+ private boolean closed = false;
+
+ public VortexRecordIterator(BufferAllocator allocator,
+ Session session,
+ DataSource dataSource,
+ Scan scan,
+ StructType schema,
+ String path) {
+ this.allocator = allocator;
+ this.session = session;
+ this.dataSource = dataSource;
+ this.scan = scan;
+ this.sparkSchema = schema;
+ this.projection = UnsafeProjection.create(schema);
+ this.path = path;
+ }
+
+ @Override
+ public boolean hasNext() {
+ if (rowIterator != null && rowIterator.hasNext()) {
+ return true;
+ }
+
+ // Advance within the current partition, then across partitions, until a
non-empty batch.
+ while (true) {
+ if (currentReader != null) {
+ if (advanceBatch()) {
+ return true;
+ }
+ // Current partition exhausted.
+ closeCurrentReader();
+ }
+ if (!scan.hasNext()) {
+ return false;
+ }
+ Partition partition = scan.next();
+ currentReader = partition.scanArrow(allocator);
+ columnVectors = null;
+ }
+ }
+
+ /**
+ * Load the next batch from the current ArrowReader (whose VectorSchemaRoot
is reused across
+ * batches, so the wrapping column vectors are built once per partition) and
set up the row
+ * iterator. Returns false when the current reader is exhausted.
+ */
+ private boolean advanceBatch() {
+ try {
+ while (currentReader.loadNextBatch()) {
+ VectorSchemaRoot root = currentReader.getVectorSchemaRoot();
+ if (columnVectors == null) {
+ columnVectors = buildColumnVectors(root);
+ }
+ currentBatch = new ColumnarBatch(columnVectors, root.getRowCount());
+ rowIterator = currentBatch.rowIterator();
+ if (rowIterator.hasNext()) {
+ return true;
+ }
+ }
+ } catch (IOException e) {
+ throw new HoodieException("Failed to read next batch from Vortex file: "
+ path, e);
+ }
+ return false;
+ }
+
+ @Override
+ public UnsafeRow next() {
+ if (!hasNext()) {
+ throw new IllegalStateException("No more records available");
+ }
+ InternalRow row = rowIterator.next();
+ return projection.apply(row).copy();
+ }
+
+ /**
+ * Wrap the requested columns of the batch as Spark {@link ColumnVector}s,
matched by name. The
+ * batch may carry more columns than requested (the scan reads all columns);
extras are ignored.
+ */
+ private ColumnVector[] buildColumnVectors(VectorSchemaRoot root) {
+ List<FieldVector> fieldVectors = root.getFieldVectors();
+ Map<String, FieldVector> byName = new HashMap<>(fieldVectors.size() * 2);
+ for (FieldVector fv : fieldVectors) {
+ byName.put(fv.getName(), fv);
+ }
+ StructField[] sparkFields = sparkSchema.fields();
+ ColumnVector[] vectors = new ColumnVector[sparkFields.length];
+ for (int i = 0; i < sparkFields.length; i++) {
+ String name = sparkFields[i].name();
+ FieldVector fv = byName.get(name);
+ if (fv == null) {
+ throw new HoodieException("Vortex batch missing expected column '" +
name
+ + "' for file: " + path + "; available columns: " +
byName.keySet());
+ }
+ vectors[i] = new ArrowColumnVector(fv);
+ }
+ return vectors;
+ }
+
+ private void closeCurrentReader() {
+ currentBatch = null;
+ columnVectors = null;
+ rowIterator = null;
+ if (currentReader != null) {
+ try {
+ currentReader.close();
+ } catch (IOException e) {
+ throw new HoodieException("Failed to close Vortex Arrow reader for
file: " + path, e);
+ } finally {
+ currentReader = null;
+ }
+ }
+ }
+
+ @Override
+ public void close() {
+ if (closed) {
+ return;
+ }
+ closed = true;
+ ArrowReader reader = currentReader;
+ currentReader = null;
+ currentBatch = null;
+ columnVectors = null;
+ rowIterator = null;
+ VortexResourceCloser.closeAll(reader, allocator, path);
+ }
+}
diff --git
a/hudi-client/hudi-spark-client/src/main/java-vortex/org/apache/hudi/io/storage/VortexResourceCloser.java
b/hudi-client/hudi-spark-client/src/main/java-vortex/org/apache/hudi/io/storage/VortexResourceCloser.java
new file mode 100644
index 000000000000..afe37170b249
--- /dev/null
+++
b/hudi-client/hudi-spark-client/src/main/java-vortex/org/apache/hudi/io/storage/VortexResourceCloser.java
@@ -0,0 +1,57 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.io.storage;
+
+import org.apache.hudi.exception.HoodieException;
+
+import org.apache.arrow.memory.BufferAllocator;
+import org.apache.arrow.vector.ipc.ArrowReader;
+
+/**
+ * Shared close logic for the Vortex reader iterator (vortex-jni 0.76.0).
Closes the current Arrow
+ * reader (which owns the exported VectorSchemaRoot / column vectors) and then
the allocator,
+ * rethrowing after the allocator is closed so a failed reader close never
leaks the allocator.
+ *
+ * <p>The Vortex Session/DataSource/Scan handles are not AutoCloseable in
0.76.0 (native cleanup is
+ * implicit), so they are not closed here.
+ */
+final class VortexResourceCloser {
+
+ private VortexResourceCloser() {
+ }
+
+ static void closeAll(ArrowReader currentReader, BufferAllocator allocator,
String path) {
+ Exception readerException = null;
+
+ if (currentReader != null) {
+ try {
+ currentReader.close();
+ } catch (Exception e) {
+ readerException = e;
+ }
+ }
+ if (allocator != null) {
+ allocator.close();
+ }
+
+ if (readerException != null) {
+ throw new HoodieException("Failed to close Vortex Arrow reader for file:
" + path, readerException);
+ }
+ }
+}
diff --git
a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/common/SparkReaderContextFactory.java
b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/common/SparkReaderContextFactory.java
index 22a21369b3c6..bb54f9cc1fc0 100644
---
a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/common/SparkReaderContextFactory.java
+++
b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/common/SparkReaderContextFactory.java
@@ -97,13 +97,17 @@ public class SparkReaderContextFactory implements
ReaderContextFactory<InternalR
SparkColumnarFileReader parquetFileReader =
sparkAdapter.createParquetFileReader(false, sqlConf, options, configs);
SparkColumnarFileReader orcFileReader = getOrcFileReader(resolver,
sqlConf, options, configs, sparkAdapter);
SparkColumnarFileReader lanceFileReader =
sparkAdapter.createLanceFileReader(false, sqlConf, options,
configs).getOrElse(null);
- baseFileReaderBroadcast = jsc.broadcast(new
MultipleColumnarFileFormatReader(parquetFileReader, orcFileReader,
lanceFileReader));
+ SparkColumnarFileReader vortexFileReader =
sparkAdapter.createVortexFileReader(false, sqlConf, options,
configs).getOrElse(null);
+ baseFileReaderBroadcast = jsc.broadcast(new
MultipleColumnarFileFormatReader(parquetFileReader, orcFileReader,
lanceFileReader, vortexFileReader));
} else if (metaClient.getTableConfig().getBaseFileFormat() ==
HoodieFileFormat.ORC) {
SparkColumnarFileReader orcFileReader = getOrcFileReader(resolver,
sqlConf, options, configs, sparkAdapter);
baseFileReaderBroadcast = jsc.broadcast(orcFileReader);
} else if (metaClient.getTableConfig().getBaseFileFormat() ==
HoodieFileFormat.LANCE) {
baseFileReaderBroadcast = jsc.broadcast(
sparkAdapter.createLanceFileReader(false, sqlConf, options,
configs).getOrElse(null));
+ } else if (metaClient.getTableConfig().getBaseFileFormat() ==
HoodieFileFormat.VORTEX) {
+ baseFileReaderBroadcast = jsc.broadcast(
+ sparkAdapter.createVortexFileReader(false, sqlConf, options,
configs).getOrElse(null));
} else {
baseFileReaderBroadcast = jsc.broadcast(
sparkAdapter.createParquetFileReader(false, sqlConf, options,
configs));
diff --git
a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/io/storage/HoodieSparkFileReaderFactory.java
b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/io/storage/HoodieSparkFileReaderFactory.java
index c9be89ab038c..c1d6bed8266f 100644
---
a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/io/storage/HoodieSparkFileReaderFactory.java
+++
b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/io/storage/HoodieSparkFileReaderFactory.java
@@ -22,6 +22,7 @@ import org.apache.hudi.common.config.HoodieConfig;
import org.apache.hudi.common.config.HoodieStorageConfig;
import org.apache.hudi.common.schema.HoodieSchema;
import org.apache.hudi.common.util.Option;
+import org.apache.hudi.common.util.ReflectionUtils;
import org.apache.hudi.core.io.storage.HoodieFileReader;
import org.apache.hudi.core.io.storage.HoodieFileReaderFactory;
import org.apache.hudi.exception.HoodieIOException;
@@ -58,6 +59,16 @@ public class HoodieSparkFileReaderFactory extends
HoodieFileReaderFactory {
return new HoodieSparkLanceReader(path, dataAllocatorSize,
metadataAllocatorSize);
}
+ @Override
+ protected HoodieFileReader newVortexFileReader(HoodieConfig hoodieConfig,
StoragePath path) {
+ long dataAllocatorSize =
hoodieConfig.getLongOrDefault(HoodieStorageConfig.VORTEX_READ_ALLOCATOR_SIZE_BYTES);
+ // Vortex sources are compiled only under JDK 17 (see the vortex profile
in this module's pom),
+ // so instantiate the reader reflectively to keep this factory compilable
on the Java 11 build.
+ return (HoodieFileReader) ReflectionUtils.loadClass(
+ "org.apache.hudi.io.storage.HoodieSparkVortexReader",
+ new Class<?>[] {StoragePath.class, long.class}, path,
dataAllocatorSize);
+ }
+
@Override
protected HoodieFileReader newHFileFileReader(HoodieConfig hoodieConfig,
StoragePath path,
diff --git
a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/io/storage/HoodieSparkFileWriterFactory.java
b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/io/storage/HoodieSparkFileWriterFactory.java
index 43264705885e..67949fee472e 100644
---
a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/io/storage/HoodieSparkFileWriterFactory.java
+++
b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/io/storage/HoodieSparkFileWriterFactory.java
@@ -25,6 +25,7 @@ import org.apache.hudi.common.engine.TaskContextSupplier;
import org.apache.hudi.common.schema.HoodieSchema;
import org.apache.hudi.common.table.HoodieTableConfig;
import org.apache.hudi.common.util.Option;
+import org.apache.hudi.common.util.ReflectionUtils;
import org.apache.hudi.common.util.collection.Pair;
import org.apache.hudi.core.io.HoodieParquetConfigInjector;
import org.apache.hudi.core.io.storage.HoodieFileWriter;
@@ -140,6 +141,27 @@ public class HoodieSparkFileWriterFactory extends
HoodieFileWriterFactory {
.build();
}
+ @Override
+ protected HoodieFileWriter newVortexFileWriter(String instantTime,
StoragePath path, HoodieConfig config, HoodieSchema schema,
+ TaskContextSupplier
taskContextSupplier) throws IOException {
+ boolean populateMetaFields =
config.getBooleanOrDefault(HoodieTableConfig.POPULATE_META_FIELDS);
+ StructType structType = HoodieInternalRowUtils.getCachedSchema(schema);
+ boolean enableBloomFilter = enableBloomFilter(populateMetaFields, config);
+ Option<BloomFilter> bloomFilter = enableBloomFilter ?
Option.of(createBloomFilter(config)) : Option.empty();
+ long maxFileSize =
config.getLongOrDefault(HoodieStorageConfig.VORTEX_MAX_FILE_SIZE);
+ long allocatorSize =
config.getLongOrDefault(HoodieStorageConfig.VORTEX_WRITE_ALLOCATOR_SIZE_BYTES);
+ long flushByteWatermark =
config.getLongOrDefault(HoodieStorageConfig.VORTEX_WRITE_FLUSH_BYTE_WATERMARK);
+
+ // Vortex sources are compiled only under JDK 17 (see the vortex profile
in this module's pom),
+ // so build the writer reflectively to keep this factory compilable on the
Java 11 build.
+ return (HoodieFileWriter) ReflectionUtils.invokeStaticMethod(
+ "org.apache.hudi.io.storage.HoodieSparkVortexWriter", "newWriter",
+ new Object[] {path, structType, instantTime, taskContextSupplier,
storage, populateMetaFields,
+ bloomFilter, maxFileSize, allocatorSize, flushByteWatermark},
+ StoragePath.class, StructType.class, String.class,
TaskContextSupplier.class, HoodieStorage.class,
+ boolean.class, Option.class, long.class, long.class, long.class);
+ }
+
private static HoodieRowParquetWriteSupport
getHoodieRowParquetWriteSupport(StorageConfiguration<?> conf, HoodieSchema
schema,
HoodieConfig config, boolean enableBloomFilter) {
Option<BloomFilter> filter = enableBloomFilter ?
Option.of(createBloomFilter(config)) : Option.empty();
diff --git
a/hudi-client/hudi-spark-client/src/main/scala/org/apache/spark/sql/hudi/MultipleColumnarFileFormatReader.scala
b/hudi-client/hudi-spark-client/src/main/scala/org/apache/spark/sql/hudi/MultipleColumnarFileFormatReader.scala
index e73378f8819e..9bce5fcb55f3 100644
---
a/hudi-client/hudi-spark-client/src/main/scala/org/apache/spark/sql/hudi/MultipleColumnarFileFormatReader.scala
+++
b/hudi-client/hudi-spark-client/src/main/scala/org/apache/spark/sql/hudi/MultipleColumnarFileFormatReader.scala
@@ -29,12 +29,12 @@ import
org.apache.spark.sql.execution.datasources.{PartitionedFile, SparkColumna
import org.apache.spark.sql.sources.Filter
import org.apache.spark.sql.types.StructType
-class MultipleColumnarFileFormatReader(parquetReader: SparkColumnarFileReader,
orcReader: SparkColumnarFileReader, lanceReader: SparkColumnarFileReader)
+class MultipleColumnarFileFormatReader(parquetReader: SparkColumnarFileReader,
orcReader: SparkColumnarFileReader, lanceReader: SparkColumnarFileReader,
vortexReader: SparkColumnarFileReader)
extends SparkColumnarFileReader with SparkAdapterSupport {
def this(parquetReader: SparkColumnarFileReader, orcReader:
SparkColumnarFileReader) = {
- // constructor for Spark versions without Lance support
- this(parquetReader, orcReader, null)
+ // constructor for Spark versions without Lance or Vortex support
+ this(parquetReader, orcReader, null, null)
}
/**
@@ -63,6 +63,11 @@ class MultipleColumnarFileFormatReader(parquetReader:
SparkColumnarFileReader, o
throw new UnsupportedOperationException("Lance format is only
supported in Spark 3.4 and above")
}
lanceReader.read(file, requiredSchema, partitionSchema,
internalSchemaOpt, filters, storageConf, tableSchemaOpt)
+ case HoodieFileFormat.VORTEX =>
+ if (vortexReader == null) {
+ throw new UnsupportedOperationException("Vortex format is only
supported in Spark 3.5 and above")
+ }
+ vortexReader.read(file, requiredSchema, partitionSchema,
internalSchemaOpt, filters, storageConf, tableSchemaOpt)
case _ =>
throw new IllegalArgumentException(s"Unsupported file format for file:
$filePath")
}
diff --git
a/hudi-client/hudi-spark-client/src/main/scala/org/apache/spark/sql/hudi/SparkAdapter.scala
b/hudi-client/hudi-spark-client/src/main/scala/org/apache/spark/sql/hudi/SparkAdapter.scala
index 2934536f4ec9..7cc32490fbfb 100644
---
a/hudi-client/hudi-spark-client/src/main/scala/org/apache/spark/sql/hudi/SparkAdapter.scala
+++
b/hudi-client/hudi-spark-client/src/main/scala/org/apache/spark/sql/hudi/SparkAdapter.scala
@@ -243,6 +243,20 @@ trait SparkAdapter extends Serializable {
options: Map[String, String],
hadoopConf: Configuration):
Option[SparkColumnarFileReader]
+ /**
+ * Get Vortex file reader
+ *
+ * @param vectorized true if vectorized reading is not prohibited due to
schema, reading mode, etc
+ * @param sqlConf the [[SQLConf]] used for the read
+ * @param options passed as a param to the file format
+ * @param hadoopConf some configs will be set for the hadoopConf
+ * @return Vortex file reader wrapped in Option; None if Vortex format is
not supported in current Spark version (i.e 3.3, 3.4)
+ */
+ def createVortexFileReader(vectorized: Boolean,
+ sqlConf: SQLConf,
+ options: Map[String, String],
+ hadoopConf: Configuration):
Option[SparkColumnarFileReader]
+
/**
* Build the [[HoodieParquetReadSupport]] for a parquet read. Spark 4.0
overrides to return
* its variant-aware subclass (variant group field reorder for the
positional converter).
diff --git
a/hudi-common/src/main/java/org/apache/hudi/common/config/HoodieStorageConfig.java
b/hudi-common/src/main/java/org/apache/hudi/common/config/HoodieStorageConfig.java
index e284e8030882..60e062da2c4d 100644
---
a/hudi-common/src/main/java/org/apache/hudi/common/config/HoodieStorageConfig.java
+++
b/hudi-common/src/main/java/org/apache/hudi/common/config/HoodieStorageConfig.java
@@ -132,6 +132,43 @@ public class HoodieStorageConfig extends HoodieConfig {
.sinceVersion("1.3.0")
.withDocumentation("Control whether to write bloom filter or not to
HFile.");
+ public static final ConfigProperty<String> VORTEX_MAX_FILE_SIZE =
ConfigProperty
+ .key("hoodie.vortex.max.file.size")
+ .defaultValue(String.valueOf(120 * 1024 * 1024))
+ .markAdvanced()
+ .withDocumentation("Target file size in bytes for Vortex base files.");
+
+ public static final ConfigProperty<String> VORTEX_WRITE_ALLOCATOR_SIZE_BYTES
= ConfigProperty
+ .key("hoodie.vortex.write.allocator.size.bytes")
+ .defaultValue(String.valueOf(256 * 1024 * 1024))
+ .markAdvanced()
+ .withDocumentation("Maximum size in bytes of the Arrow child allocator
used by the Vortex "
+ + "writer for buffering in-flight batch data. Must be large enough
that the Arrow "
+ + "buffer's power-of-2 doubling growth never requests a buffer
exceeding this cap, "
+ + "otherwise writes fail with OutOfMemoryException.");
+
+ public static final ConfigProperty<String> VORTEX_WRITE_FLUSH_BYTE_WATERMARK
= ConfigProperty
+ .key("hoodie.vortex.write.flush.byte.watermark")
+ .defaultValue(String.valueOf(96 * 1024 * 1024))
+ .markAdvanced()
+ .withDocumentation("Byte-size watermark on the Vortex writer's in-flight
Arrow buffers; "
+ + "the writer flushes the current batch when the sum of FieldVector
buffer sizes "
+ + "reaches this value, in addition to the row-count batch
threshold.");
+
+ public static final ConfigProperty<String> VORTEX_READ_ALLOCATOR_SIZE_BYTES
= ConfigProperty
+ .key("hoodie.vortex.read.allocator.size.bytes")
+ .defaultValue(String.valueOf(256 * 1024 * 1024))
+ .markAdvanced()
+ .withDocumentation("Maximum size in bytes of the Arrow child allocator
used by the Vortex "
+ + "reader for per-read data buffers.");
+
+ public static final ConfigProperty<String>
VORTEX_READ_METADATA_ALLOCATOR_SIZE_BYTES = ConfigProperty
+ .key("hoodie.vortex.read.metadata.allocator.size.bytes")
+ .defaultValue(String.valueOf(8 * 1024 * 1024))
+ .markAdvanced()
+ .withDocumentation("Maximum size in bytes of the Arrow child allocator
used by the Vortex "
+ + "reader for footer/metadata operations (schema, bloom filter).");
+
public static final ConfigProperty<Boolean> HFILE_WRITER_TO_ALLOW_DUPLICATES
= ConfigProperty
.key("hoodie.hfile.writes.allow.duplicates")
.defaultValue(false)
@@ -632,6 +669,11 @@ public class HoodieStorageConfig extends HoodieConfig {
return this;
}
+ public Builder vortexMaxFileSize(long maxFileSize) {
+ storageConfig.setValue(VORTEX_MAX_FILE_SIZE,
String.valueOf(maxFileSize));
+ return this;
+ }
+
public Builder orcStripeSize(int orcStripeSize) {
storageConfig.setValue(ORC_STRIPE_SIZE, String.valueOf(orcStripeSize));
return this;
diff --git
a/hudi-common/src/main/java/org/apache/hudi/common/model/HoodieFileFormat.java
b/hudi-common/src/main/java/org/apache/hudi/common/model/HoodieFileFormat.java
index 8410ad486dea..8961689ec9c8 100644
---
a/hudi-common/src/main/java/org/apache/hudi/common/model/HoodieFileFormat.java
+++
b/hudi-common/src/main/java/org/apache/hudi/common/model/HoodieFileFormat.java
@@ -53,12 +53,20 @@ public enum HoodieFileFormat {
@EnumFieldDescription("Lance is a modern columnar data format optimized for
random access patterns "
+ "and designed for ML and AI workloads")
- LANCE(".lance");
+ LANCE(".lance"),
+
+ @EnumFieldDescription("Vortex is a next-generation columnar data format with
cascading compression "
+ + "optimized for fast random access and analytical scans")
+ VORTEX(".vortex");
public static final String LANCE_UNSUPPORTED_ERROR_MSG =
"Lance base file format is not supported by this reader or writer path. "
+ "Please use an engine-specific Lance reader/writer, or use
Parquet, ORC, or HFile.";
+ public static final String VORTEX_SPARK_ONLY_ERROR_MSG =
+ "Vortex base file format is currently only supported with the Spark
engine. "
+ + "Please use Parquet, ORC, or HFile for non-Spark engines (Flink,
Hive, Presto, Trino).";
+
public static final Set<String> BASE_FILE_EXTENSIONS =
Arrays.stream(HoodieFileFormat.values())
.map(HoodieFileFormat::getFileExtension)
.filter(x -> !x.equals(HoodieFileFormat.HOODIE_LOG.getFileExtension()))
diff --git
a/hudi-common/src/main/java/org/apache/hudi/common/util/VortexUtils.java
b/hudi-common/src/main/java/org/apache/hudi/common/util/VortexUtils.java
new file mode 100644
index 000000000000..e51ab8910a4c
--- /dev/null
+++ b/hudi-common/src/main/java/org/apache/hudi/common/util/VortexUtils.java
@@ -0,0 +1,203 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.hudi.common.util;
+
+import org.apache.hudi.common.config.HoodieReaderConfig;
+import org.apache.hudi.common.model.HoodieFileFormat;
+import org.apache.hudi.common.model.HoodieKey;
+import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.common.schema.HoodieSchema;
+import org.apache.hudi.common.util.collection.ClosableIterator;
+import org.apache.hudi.common.util.collection.CloseableMappingIterator;
+import org.apache.hudi.common.util.collection.Pair;
+import org.apache.hudi.core.io.storage.HoodieFileReader;
+import org.apache.hudi.core.io.storage.HoodieIOFactory;
+import org.apache.hudi.exception.HoodieIOException;
+import org.apache.hudi.keygen.BaseKeyGenerator;
+import org.apache.hudi.metadata.HoodieIndexVersion;
+import org.apache.hudi.metadata.stats.HoodieColumnRangeMetadata;
+import org.apache.hudi.storage.HoodieStorage;
+import org.apache.hudi.storage.StoragePath;
+
+import org.apache.avro.generic.GenericRecord;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.Set;
+import java.util.concurrent.atomic.AtomicLong;
+
+public class VortexUtils extends FileFormatUtils {
+
+ @Override
+ public ClosableIterator<Pair<HoodieKey, Long>>
fetchRecordKeysWithPositions(HoodieStorage storage, StoragePath filePath) {
+ return fetchRecordKeysWithPositions(storage, filePath, Option.empty(),
Option.empty());
+ }
+
+ @Override
+ public ClosableIterator<Pair<HoodieKey, Long>>
fetchRecordKeysWithPositions(HoodieStorage storage,
+
StoragePath filePath,
+
Option<BaseKeyGenerator> keyGeneratorOpt,
+
Option<String> partitionPath) {
+ AtomicLong position = new AtomicLong(0);
+ return new CloseableMappingIterator<>(
+ getHoodieKeyIterator(storage, filePath, keyGeneratorOpt,
partitionPath),
+ key -> Pair.of(key, position.getAndIncrement()));
+ }
+
+ @Override
+ public ClosableIterator<HoodieKey> getHoodieKeyIterator(HoodieStorage
storage, StoragePath filePath) {
+ return getHoodieKeyIterator(storage, filePath, Option.empty(),
Option.empty());
+ }
+
+ @Override
+ public ClosableIterator<HoodieKey> getHoodieKeyIterator(HoodieStorage
storage,
+ StoragePath
filePath,
+
Option<BaseKeyGenerator> keyGeneratorOpt,
+ Option<String>
partitionPath) {
+ try {
+ HoodieSchema keySchema = getKeyIteratorSchema(storage, filePath,
keyGeneratorOpt, partitionPath);
+ HoodieFileReader reader = HoodieIOFactory.getIOFactory(storage)
+ .getReaderFactory(HoodieRecord.HoodieRecordType.SPARK)
+ .getFileReader(new HoodieReaderConfig(), filePath,
HoodieFileFormat.VORTEX);
+ ClosableIterator<HoodieRecord> recordIterator =
reader.getRecordIterator(keySchema);
+
+ return new CloseableMappingIterator<>(
+ recordIterator,
+ record -> {
+ String recordKey;
+ if (keyGeneratorOpt.isPresent()) {
+ recordKey = record.getRecordKey(keySchema, keyGeneratorOpt);
+ } else {
+ recordKey = record.getRecordKey(keySchema,
HoodieRecord.RECORD_KEY_METADATA_FIELD);
+ }
+ String partitionPathValue = partitionPath.orElseGet(() ->
+ (String) record.getColumnValueAsJava(keySchema,
HoodieRecord.PARTITION_PATH_METADATA_FIELD,
+ CollectionUtils.emptyProps()));
+ return new HoodieKey(recordKey, partitionPathValue);
+ }
+ );
+ } catch (IOException e) {
+ throw new HoodieIOException("Failed to read from Vortex file " +
filePath, e);
+ }
+ }
+
+ @Override
+ public HoodieSchema readSchema(HoodieStorage storage, StoragePath filePath) {
+ try (HoodieFileReader fileReader =
+ HoodieIOFactory.getIOFactory(storage)
+ .getReaderFactory(HoodieRecord.HoodieRecordType.SPARK)
+ .getFileReader(
+ ConfigUtils.DEFAULT_HUDI_CONFIG_FOR_READER,
+ filePath,
+ HoodieFileFormat.VORTEX)) {
+ return fileReader.getSchema();
+ } catch (IOException e) {
+ throw new HoodieIOException("Failed to read schema from Vortex file", e);
+ }
+ }
+
+ @Override
+ public HoodieFileFormat getFormat() {
+ return HoodieFileFormat.VORTEX;
+ }
+
+ @Override
+ public List<GenericRecord> readAvroRecords(HoodieStorage storage,
StoragePath filePath) {
+ throw new UnsupportedOperationException("readAvroRecords is not yet
supported for Vortex format");
+ }
+
+ @Override
+ public List<GenericRecord> readAvroRecords(HoodieStorage storage,
StoragePath filePath, HoodieSchema schema) {
+ throw new UnsupportedOperationException("readAvroRecords with schema is
not yet supported for Vortex format");
+ }
+
+ @Override
+ public Map<String, String> readFooter(HoodieStorage storage, boolean
required, StoragePath filePath, String... footerNames) {
+ throw new UnsupportedOperationException("readFooter is not yet supported
for Vortex format");
+ }
+
+ @Override
+ public long getRowCount(HoodieStorage storage, StoragePath filePath) {
+ try (HoodieFileReader fileReader =
+ HoodieIOFactory.getIOFactory(storage)
+ .getReaderFactory(HoodieRecord.HoodieRecordType.SPARK)
+ .getFileReader(
+ ConfigUtils.DEFAULT_HUDI_CONFIG_FOR_READER,
+ filePath,
+ HoodieFileFormat.VORTEX)) {
+ return fileReader.getTotalRecords();
+ } catch (IOException e) {
+ throw new HoodieIOException("Failed to get row count from Vortex file",
e);
+ }
+ }
+
+ @Override
+ public Set<Pair<String, Long>> filterRowKeys(HoodieStorage storage,
StoragePath filePath, Set<String> filterKeys) {
+ try (HoodieFileReader fileReader =
+ HoodieIOFactory.getIOFactory(storage)
+ .getReaderFactory(HoodieRecord.HoodieRecordType.SPARK)
+ .getFileReader(
+ ConfigUtils.DEFAULT_HUDI_CONFIG_FOR_READER,
+ filePath,
+ HoodieFileFormat.VORTEX)) {
+ return fileReader.filterRowKeys(filterKeys);
+ } catch (IOException e) {
+ throw new HoodieIOException("Failed to read filter keys from Vortex
file", e);
+ }
+ }
+
+ @Override
+ public List<HoodieColumnRangeMetadata<Comparable>>
readColumnStatsFromMetadata(HoodieStorage storage,
+
StoragePath filePath,
+
List<String> columnList,
+
HoodieIndexVersion indexVersion) {
+ throw new UnsupportedOperationException("readColumnStatsFromMetadata is
not yet supported for Vortex format");
+ }
+
+ @Override
+ public void writeMetaFile(HoodieStorage storage, StoragePath filePath,
Properties props) throws IOException {
+ throw new UnsupportedOperationException("writeMetaFile is not yet
supported for Vortex format");
+ }
+
+ @Override
+ public ByteArrayOutputStream serializeRecordsToLogBlock(HoodieStorage
storage,
+ List<HoodieRecord>
records,
+ HoodieSchema
writerSchema,
+ HoodieSchema
readerSchema,
+ String keyFieldName,
+ Map<String, String>
paramsMap) throws IOException {
+ throw new UnsupportedOperationException("serializeRecordsToLogBlock is not
yet supported for Vortex format");
+ }
+
+ @Override
+ public Pair<ByteArrayOutputStream, Object>
serializeRecordsToLogBlock(HoodieStorage storage,
+
Iterator<HoodieRecord> records,
+
HoodieRecord.HoodieRecordType recordType,
+
HoodieSchema writerSchema,
+
HoodieSchema readerSchema,
+ String
keyFieldName,
+
Map<String, String> paramsMap) throws IOException {
+ throw new UnsupportedOperationException("serializeRecordsToLogBlock with
iterator is not yet supported for Vortex format");
+ }
+}
diff --git
a/hudi-common/src/main/java/org/apache/hudi/core/io/storage/HoodieFileReaderFactory.java
b/hudi-common/src/main/java/org/apache/hudi/core/io/storage/HoodieFileReaderFactory.java
index 1be9b4f34282..ba0481171ae7 100644
---
a/hudi-common/src/main/java/org/apache/hudi/core/io/storage/HoodieFileReaderFactory.java
+++
b/hudi-common/src/main/java/org/apache/hudi/core/io/storage/HoodieFileReaderFactory.java
@@ -33,6 +33,7 @@ import static
org.apache.hudi.common.model.HoodieFileFormat.HFILE;
import static org.apache.hudi.common.model.HoodieFileFormat.LANCE;
import static org.apache.hudi.common.model.HoodieFileFormat.ORC;
import static org.apache.hudi.common.model.HoodieFileFormat.PARQUET;
+import static org.apache.hudi.common.model.HoodieFileFormat.VORTEX;
/**
* Factory methods to create Hudi file reader.
@@ -58,6 +59,9 @@ public class HoodieFileReaderFactory {
if (LANCE.getFileExtension().equals(extension)) {
return getFileReader(hoodieConfig, path, LANCE, Option.empty());
}
+ if (VORTEX.getFileExtension().equals(extension)) {
+ return getFileReader(hoodieConfig, path, VORTEX, Option.empty());
+ }
throw new UnsupportedOperationException(extension + " format not supported
yet.");
}
@@ -77,6 +81,8 @@ public class HoodieFileReaderFactory {
return newOrcFileReader(path);
case LANCE:
return newLanceFileReader(hoodieConfig, path);
+ case VORTEX:
+ return newVortexFileReader(hoodieConfig, path);
default:
throw new UnsupportedOperationException(format + " format not
supported yet.");
}
@@ -137,6 +143,10 @@ public class HoodieFileReaderFactory {
throw new
UnsupportedOperationException(HoodieFileFormat.LANCE_UNSUPPORTED_ERROR_MSG);
}
+ protected HoodieFileReader newVortexFileReader(HoodieConfig hoodieConfig,
StoragePath path) {
+ throw new
UnsupportedOperationException(HoodieFileFormat.VORTEX_SPARK_ONLY_ERROR_MSG);
+ }
+
public HoodieFileReader newBootstrapFileReader(HoodieFileReader
skeletonFileReader,
HoodieFileReader
dataFileReader,
Option<String[]>
partitionFields,
diff --git
a/hudi-common/src/main/java/org/apache/hudi/core/io/storage/HoodieFileWriterFactory.java
b/hudi-common/src/main/java/org/apache/hudi/core/io/storage/HoodieFileWriterFactory.java
index 7c8499e42c3d..2ac944085986 100644
---
a/hudi-common/src/main/java/org/apache/hudi/core/io/storage/HoodieFileWriterFactory.java
+++
b/hudi-common/src/main/java/org/apache/hudi/core/io/storage/HoodieFileWriterFactory.java
@@ -37,6 +37,7 @@ import static
org.apache.hudi.common.model.HoodieFileFormat.HFILE;
import static org.apache.hudi.common.model.HoodieFileFormat.LANCE;
import static org.apache.hudi.common.model.HoodieFileFormat.ORC;
import static org.apache.hudi.common.model.HoodieFileFormat.PARQUET;
+import static org.apache.hudi.common.model.HoodieFileFormat.VORTEX;
public class HoodieFileWriterFactory {
protected final HoodieStorage storage;
@@ -75,6 +76,9 @@ public class HoodieFileWriterFactory {
if (LANCE.getFileExtension().equals(extension)) {
return newLanceFileWriter(instantTime, path, config, schema,
taskContextSupplier);
}
+ if (VORTEX.getFileExtension().equals(extension)) {
+ return newVortexFileWriter(instantTime, path, config, schema,
taskContextSupplier);
+ }
throw new UnsupportedOperationException(extension + " format not supported
yet.");
}
@@ -117,6 +121,12 @@ public class HoodieFileWriterFactory {
throw new
UnsupportedOperationException(HoodieFileFormat.LANCE_UNSUPPORTED_ERROR_MSG);
}
+ protected HoodieFileWriter newVortexFileWriter(
+ String instantTime, StoragePath path, HoodieConfig config, HoodieSchema
schema,
+ TaskContextSupplier taskContextSupplier) throws IOException {
+ throw new
UnsupportedOperationException(HoodieFileFormat.VORTEX_SPARK_ONLY_ERROR_MSG);
+ }
+
public static BloomFilter createBloomFilter(HoodieConfig config) {
return BloomFilterFactory.createBloomFilter(
config.getIntOrDefault(HoodieStorageConfig.BLOOM_FILTER_NUM_ENTRIES_VALUE),
diff --git
a/hudi-common/src/main/java/org/apache/hudi/core/io/storage/HoodieIOFactory.java
b/hudi-common/src/main/java/org/apache/hudi/core/io/storage/HoodieIOFactory.java
index 57dcc753c9d0..8e52f781c5b4 100644
---
a/hudi-common/src/main/java/org/apache/hudi/core/io/storage/HoodieIOFactory.java
+++
b/hudi-common/src/main/java/org/apache/hudi/core/io/storage/HoodieIOFactory.java
@@ -114,6 +114,8 @@ public abstract class HoodieIOFactory {
return getFileFormatUtils(HoodieFileFormat.HFILE);
} else if
(path.getFileExtension().equals(HoodieFileFormat.LANCE.getFileExtension())) {
return getFileFormatUtils(HoodieFileFormat.LANCE);
+ } else if
(path.getFileExtension().equals(HoodieFileFormat.VORTEX.getFileExtension())) {
+ return getFileFormatUtils(HoodieFileFormat.VORTEX);
}
throw new UnsupportedOperationException("The format for file " + path + "
is not supported yet.");
}
diff --git a/hudi-hadoop-common/pom.xml b/hudi-hadoop-common/pom.xml
index 7e6b8615fe1c..c0b106e982ff 100644
--- a/hudi-hadoop-common/pom.xml
+++ b/hudi-hadoop-common/pom.xml
@@ -164,4 +164,48 @@
<artifactId>lance-core</artifactId>
</dependency>
</dependencies>
+
+ <profiles>
+ <!-- Vortex requires Java 17: dev.vortex:vortex-jni ships Java 17 bytecode
(class file major
+ version 61), which javac 11 (used by the Java 11 CI jobs) cannot
read. Compile the Vortex
+ sources under src/main/java-vortex and pull in vortex-jni only when
building on JDK 17;
+ Java 8/11 builds skip the format entirely. -->
+ <profile>
+ <id>vortex</id>
+ <activation>
+ <jdk>[17,)</jdk>
+ </activation>
+ <dependencies>
+ <dependency>
+ <groupId>dev.vortex</groupId>
+ <artifactId>vortex-jni</artifactId>
+ </dependency>
+ </dependencies>
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.codehaus.mojo</groupId>
+ <artifactId>build-helper-maven-plugin</artifactId>
+ <executions>
+ <execution>
+ <id>add-vortex-source</id>
+ <phase>generate-sources</phase>
+ <goals>
+ <goal>add-source</goal>
+ </goals>
+ <configuration>
+ <!-- Vortex is only supported on Spark 3.5+/4.x; skip the
source on Spark 3.3/3.4
+ (vortex.skip.tests=true) where its Spark 3.5+ APIs are
unavailable. -->
+ <skipAddSource>${vortex.skip.tests}</skipAddSource>
+ <sources>
+ <source>src/main/java-vortex</source>
+ </sources>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
+ </plugins>
+ </build>
+ </profile>
+ </profiles>
</project>
diff --git
a/hudi-hadoop-common/src/main/java-vortex/org/apache/hudi/io/vortex/HoodieBaseVortexWriter.java
b/hudi-hadoop-common/src/main/java-vortex/org/apache/hudi/io/vortex/HoodieBaseVortexWriter.java
new file mode 100644
index 000000000000..c73d629152ec
--- /dev/null
+++
b/hudi-hadoop-common/src/main/java-vortex/org/apache/hudi/io/vortex/HoodieBaseVortexWriter.java
@@ -0,0 +1,345 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.hudi.io.vortex;
+
+import org.apache.hudi.common.avro.HoodieBloomFilterWriteSupport;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.exception.HoodieException;
+import org.apache.hudi.io.memory.HoodieArrowAllocator;
+import org.apache.hudi.storage.StoragePath;
+
+import dev.vortex.api.Session;
+import dev.vortex.api.VortexWriter;
+import lombok.AccessLevel;
+import lombok.Getter;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.arrow.c.ArrowArray;
+import org.apache.arrow.c.ArrowSchema;
+import org.apache.arrow.c.Data;
+import org.apache.arrow.memory.BufferAllocator;
+import org.apache.arrow.memory.RootAllocator;
+import org.apache.arrow.vector.FieldVector;
+import org.apache.arrow.vector.VectorSchemaRoot;
+import org.apache.arrow.vector.types.pojo.Schema;
+
+import javax.annotation.concurrent.NotThreadSafe;
+
+import java.io.Closeable;
+import java.io.File;
+import java.io.IOException;
+import java.net.URI;
+import java.util.Collections;
+
+/**
+ * Base class for Hudi Vortex file writers supporting different record types.
+ *
+ * This class handles common Vortex file operations including:
+ * - {@link VortexWriter} lifecycle management (vortex-jni {@code
dev.vortex.api})
+ * - BufferAllocator management
+ * - Record buffering and batch flushing. Each flushed batch is exported
through the Arrow C Data
+ * Interface and handed to {@link VortexWriter#writeBatch(long, long)};
because vortex copies the
+ * batch without invoking the C release callback, {@link
ArrowArray#release()} is called after each
+ * write so the exported buffers are freed.
+ * - File size checks
+ *
+ * Subclasses must implement type-specific conversion to Arrow format and
provide the Arrow schema.
+ *
+ * <p>NOTE: vortex-jni 0.76.0 exposes no file-level custom-metadata API, so
Hudi's bloom filter and
+ * min/max record-key footers cannot be persisted into the Vortex file yet.
Record keys are still
+ * tracked via {@link #bloomFilterWriteSupportOpt} but the finalized metadata
is not written.
+ * TODO(#18623): persist bloom/min-max footers once Vortex supports file
metadata.
+ *
+ * @param <R> The record type (e.g., GenericRecord, InternalRow)
+ */
+@Slf4j
+@NotThreadSafe
+public abstract class HoodieBaseVortexWriter<R, K extends Comparable<K>>
implements Closeable {
+ protected static final int DEFAULT_BATCH_SIZE = 1000;
+ private final StoragePath path;
+ private final BufferAllocator allocator;
+ private final int batchSize;
+ private final long flushByteWatermark;
+ @Getter(value = AccessLevel.PROTECTED)
+ private long writtenRecordCount = 0;
+ private long totalFlushedDataSize = 0;
+ private int currentBatchSize = 0;
+ private VectorSchemaRoot root;
+ private ArrowWriter<R> arrowWriter;
+ protected final Option<HoodieBloomFilterWriteSupport<K>>
bloomFilterWriteSupportOpt;
+
+ // Held for the writer's lifetime. session/nativeAllocator back the native
writer; nativeAllocator
+ // is a dedicated allocator isolated from the strict data allocator (see
initializeWriter).
+ private Session session;
+ private VortexWriter writer;
+ private BufferAllocator nativeAllocator;
+ private ArrowSchema exportedSchema;
+ private long exportedSchemaAddr;
+
+ /**
+ * Constructor for base Vortex writer.
+ *
+ * @param path Path where Vortex file will be written
+ * @param batchSize Row-count threshold; the current batch is flushed when
this many records have been buffered
+ * @param allocatorSize Maximum bytes the per-writer Arrow child allocator
may hold at once
+ * @param flushByteWatermark Byte-size threshold; the current batch is
flushed when the sum of in-flight
+ * FieldVector buffer sizes reaches this value
+ * @param bloomFilterWriteSupportOpt Optional bloom filter write support for
record key tracking
+ */
+ protected HoodieBaseVortexWriter(StoragePath path, int batchSize, long
allocatorSize, long flushByteWatermark,
+ Option<HoodieBloomFilterWriteSupport<K>>
bloomFilterWriteSupportOpt) {
+ this.path = path;
+ this.allocator = HoodieArrowAllocator.newChildAllocator(
+ getClass().getSimpleName() + "-data-" + path.getName(), allocatorSize);
+ this.batchSize = batchSize;
+ this.flushByteWatermark = flushByteWatermark;
+ this.bloomFilterWriteSupportOpt = bloomFilterWriteSupportOpt;
+ }
+
+ /**
+ * Create and initialize the arrow writer for writing records to
VectorSchemaRoot.
+ * Called once during lazy initialization when the first record is written.
+ *
+ * @param root The VectorSchemaRoot to write into
+ * @return An arrow writer implementation that writes records of type R to
the root
+ */
+ protected abstract ArrowWriter<R> createArrowWriter(VectorSchemaRoot root);
+
+ /**
+ * Get the Arrow schema for this writer. It is passed directly to {@link
VortexWriter} and each
+ * written batch must conform to it.
+ *
+ * @return Arrow schema
+ */
+ protected abstract Schema getArrowSchema();
+
+ /**
+ * Write a single record. Records are buffered and flushed in batches.
+ *
+ * @param record Record to write
+ * @throws IOException if write fails
+ */
+ public void write(R record) throws IOException {
+ // Lazy initialization on first write
+ if (writer == null) {
+ initializeWriter();
+ }
+ if (root == null) {
+ root = VectorSchemaRoot.create(getArrowSchema(), allocator);
+ }
+ if (arrowWriter == null) {
+ arrowWriter = createArrowWriter(root);
+ }
+
+ // Reset arrow writer at the start of each new batch
+ if (currentBatchSize == 0) {
+ arrowWriter.reset();
+ }
+
+ arrowWriter.write(record);
+ currentBatchSize++;
+ writtenRecordCount++;
+
+ // Flush when row-count batch is full OR in-flight Arrow buffers cross the
byte watermark.
+ if (currentBatchSize >= batchSize || currentBufferBytes() >=
flushByteWatermark) {
+ flushBatch();
+ }
+ }
+
+ /**
+ * Bytes currently held by the writer's Arrow child allocator.
+ */
+ private long currentBufferBytes() {
+ return allocator.getAllocatedMemory();
+ }
+
+ /**
+ * Close the writer, flushing any remaining buffered records.
+ *
+ * @throws IOException if close fails
+ */
+ @Override
+ public void close() throws IOException {
+ Exception primaryException = null;
+
+ // 1. Flush remaining records
+ try {
+ if (currentBatchSize > 0) {
+ flushBatch();
+ }
+
+ // Ensure the file is created even if no data was written, by emitting a
single empty batch.
+ if (writer == null) {
+ initializeWriter();
+ try (VectorSchemaRoot emptyRoot =
VectorSchemaRoot.create(getArrowSchema(), allocator)) {
+ emptyRoot.setRowCount(0);
+ writeBatchFfi(emptyRoot);
+ }
+ }
+ } catch (Exception e) {
+ primaryException = e;
+ }
+
+ // Close Vortex writer (finalizes the file)
+ if (writer != null) {
+ try {
+ writer.close();
+ } catch (Exception e) {
+ primaryException = addSuppressed(primaryException, e);
+ }
+ }
+
+ // Close the exported schema handle.
+ if (exportedSchema != null) {
+ try {
+ exportedSchema.release();
+ exportedSchema.close();
+ } catch (Exception e) {
+ primaryException = addSuppressed(primaryException, e);
+ }
+ }
+
+ // Close VectorSchemaRoot
+ if (root != null) {
+ try {
+ root.close();
+ } catch (Exception e) {
+ primaryException = addSuppressed(primaryException, e);
+ }
+ }
+
+ // Close the strict data allocator (per-batch exports are released, so
this should be clean).
+ try {
+ allocator.close();
+ } catch (Exception e) {
+ primaryException = addSuppressed(primaryException, e);
+ }
+
+ // Close the native allocator last. vortex-jni 0.76.0's
VortexWriter.create exports the file
+ // schema into this allocator and relies on GC/Cleaner-based cleanup
rather than freeing it on
+ // close(), so a small one-time amount can still be outstanding here;
tolerate it (it is freed
+ // when the native writer is reclaimed) rather than failing the whole
write.
+ // TODO(#18623): drop this once vortex-jni frees the create() schema
export on writer close.
+ if (nativeAllocator != null) {
+ try {
+ nativeAllocator.close();
+ } catch (IllegalStateException e) {
+ log.warn("Vortex native allocator retained memory at close for {}
(vortex-jni create() "
+ + "schema export, reclaimed on GC): {}", path, e.getMessage());
+ }
+ }
+
+ if (primaryException != null) {
+ throw new HoodieException("Failed to close Vortex writer: " + path,
primaryException);
+ }
+ }
+
+ private static Exception addSuppressed(Exception primary, Exception e) {
+ if (primary == null) {
+ return e;
+ }
+ primary.addSuppressed(e);
+ return primary;
+ }
+
+ /**
+ * Returns the estimated data size in bytes, including both flushed batches
and the current
+ * in-progress batch.
+ */
+ protected long getDataSize() {
+ long currentBufferSize = currentBatchSize > 0 ?
allocator.getAllocatedMemory() : 0;
+ return totalFlushedDataSize + currentBufferSize;
+ }
+
+ /**
+ * Flush buffered records to Vortex file.
+ */
+ private void flushBatch() throws IOException {
+ if (currentBatchSize == 0) {
+ return;
+ }
+
+ arrowWriter.finishBatch();
+
+ for (FieldVector vector : root.getFieldVectors()) {
+ totalFlushedDataSize += vector.getBufferSize();
+ }
+
+ writeBatchFfi(root);
+
+ // Release Arrow buffers so capacity does not accumulate across batches.
+ root.close();
+ root = null;
+ arrowWriter = null;
+
+ currentBatchSize = 0;
+ }
+
+ /**
+ * Export a VectorSchemaRoot's array through the Arrow C Data Interface and
write it to the Vortex
+ * file. The array is exported from the strict data allocator and released
after the write (vortex
+ * copies the batch but does not invoke the C release callback), so no
data-allocator memory leaks.
+ */
+ private void writeBatchFfi(VectorSchemaRoot batch) throws IOException {
+ try (ArrowArray cArray = ArrowArray.allocateNew(allocator)) {
+ Data.exportVectorSchemaRoot(allocator, batch, null, cArray);
+ writer.writeBatch(cArray.memoryAddress(), exportedSchemaAddr);
+ cArray.release();
+ }
+ }
+
+ /**
+ * Initialize the {@link VortexWriter} (lazy initialization). The writer and
its exported schema
+ * use a dedicated {@link #nativeAllocator} so the native side's
non-deterministic cleanup does
+ * not pollute the strict per-batch data allocator.
+ */
+ private void initializeWriter() throws IOException {
+ session = Session.create();
+ nativeAllocator = new RootAllocator();
+ Schema arrowSchema = getArrowSchema();
+ writer = VortexWriter.create(session, toVortexUri(path), arrowSchema,
Collections.emptyMap(), nativeAllocator);
+ exportedSchema = ArrowSchema.allocateNew(nativeAllocator);
+ Data.exportSchema(nativeAllocator, arrowSchema, null, exportedSchema);
+ exportedSchemaAddr = exportedSchema.memoryAddress();
+ }
+
+ /**
+ * Vortex requires a well-formed URL. Hudi {@link StoragePath}s for remote
stores already carry a
+ * scheme (s3://, hdfs://, ...); local paths may not, so qualify them as
{@code file://} URIs.
+ */
+ private static String toVortexUri(StoragePath path) {
+ URI uri = path.toUri();
+ if (uri.getScheme() == null) {
+ return new File(uri.getPath()).toURI().toString();
+ }
+ return uri.toString();
+ }
+
+ /**
+ * Arrow writer interface for writing records of type T to a
VectorSchemaRoot.
+ * @param <T> the record type
+ */
+ protected interface ArrowWriter<T> {
+ void write(T row) throws IOException;
+
+ void finishBatch() throws IOException;
+
+ void reset();
+ }
+}
diff --git
a/hudi-hadoop-common/src/main/java/org/apache/hudi/io/storage/hadoop/HoodieHadoopIOFactory.java
b/hudi-hadoop-common/src/main/java/org/apache/hudi/io/storage/hadoop/HoodieHadoopIOFactory.java
index 0dcc6dacb823..550203002354 100644
---
a/hudi-hadoop-common/src/main/java/org/apache/hudi/io/storage/hadoop/HoodieHadoopIOFactory.java
+++
b/hudi-hadoop-common/src/main/java/org/apache/hudi/io/storage/hadoop/HoodieHadoopIOFactory.java
@@ -28,6 +28,7 @@ import org.apache.hudi.common.util.LanceUtils;
import org.apache.hudi.common.util.OrcUtils;
import org.apache.hudi.common.util.ParquetUtils;
import org.apache.hudi.common.util.ReflectionUtils;
+import org.apache.hudi.common.util.VortexUtils;
import org.apache.hudi.core.io.storage.HoodieFileReaderFactory;
import org.apache.hudi.core.io.storage.HoodieFileWriterFactory;
import org.apache.hudi.core.io.storage.HoodieIOFactory;
@@ -114,6 +115,8 @@ public class HoodieHadoopIOFactory extends HoodieIOFactory {
return new HFileUtils();
case LANCE:
return new LanceUtils();
+ case VORTEX:
+ return new VortexUtils();
default:
throw new UnsupportedOperationException(fileFormat.name() + " format
not supported yet.");
}
diff --git
a/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/HiveHoodieReaderContext.java
b/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/HiveHoodieReaderContext.java
index ad887526441d..93723e99a9ea 100644
---
a/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/HiveHoodieReaderContext.java
+++
b/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/HiveHoodieReaderContext.java
@@ -132,6 +132,9 @@ public class HiveHoodieReaderContext extends
HoodieReaderContext<ArrayWritable>
if
(filePath.toString().endsWith(HoodieFileFormat.LANCE.getFileExtension())) {
throw new
UnsupportedOperationException(HoodieFileFormat.LANCE_UNSUPPORTED_ERROR_MSG);
}
+ if
(filePath.toString().endsWith(HoodieFileFormat.VORTEX.getFileExtension())) {
+ throw new
UnsupportedOperationException(HoodieFileFormat.VORTEX_SPARK_ONLY_ERROR_MSG);
+ }
// mdt file schema irregular and does not work with this logic. Also, log
file evolution is handled inside the log block
boolean isParquetOrOrc =
filePath.getFileExtension().equals(HoodieFileFormat.PARQUET.getFileExtension())
||
filePath.getFileExtension().equals(HoodieFileFormat.ORC.getFileExtension());
diff --git
a/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/HoodieVortexInputFormat.java
b/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/HoodieVortexInputFormat.java
new file mode 100644
index 000000000000..6f59b9caf194
--- /dev/null
+++
b/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/HoodieVortexInputFormat.java
@@ -0,0 +1,67 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.hadoop;
+
+import org.apache.hudi.common.table.timeline.HoodieTimeline;
+import org.apache.hudi.hadoop.utils.HoodieInputFormatUtils;
+
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.io.ArrayWritable;
+import org.apache.hadoop.io.NullWritable;
+import org.apache.hadoop.mapred.InputSplit;
+import org.apache.hadoop.mapred.JobConf;
+import org.apache.hadoop.mapred.RecordReader;
+import org.apache.hadoop.mapred.Reporter;
+
+import java.io.IOException;
+
+/**
+ * HoodieInputFormat for HUDI datasets which store data in Vortex base file
format.
+ * <p>
+ * This class is required for catalog/metastore registration during CREATE
TABLE operations.
+ * <p>
+ * TODO(#18623): Vortex reading through Hive InputFormat is not yet supported.
When support is
+ * added, this should route through {@link
HoodieFileGroupReaderBasedRecordReader} (like
+ * {@link HoodieParquetInputFormat} does) instead of a standalone record
reader,
+ * to get MOR log merging, schema evolution, and bootstrap support for free.
+ *
+ * @see <a href="https://github.com/apache/hudi/issues/18623";>#18623</a>
+ */
+@UseFileSplitsFromInputFormat
+public class HoodieVortexInputFormat extends HoodieCopyOnWriteTableInputFormat
{
+
+ protected HoodieTimeline filterInstantsTimeline(HoodieTimeline timeline) {
+ return HoodieInputFormatUtils.filterInstantsTimeline(timeline);
+ }
+
+ @Override
+ public RecordReader<NullWritable, ArrayWritable> getRecordReader(final
InputSplit split, final JobConf job,
+ final Reporter reporter) throws IOException {
+ throw new UnsupportedOperationException(
+ "Vortex reading through Hive InputFormat is not yet supported. "
+ + "Use the Spark datasource path (spark.read.format(\"hudi\")) to
read Vortex tables.");
+ }
+
+ @Override
+ protected boolean isSplitable(FileSystem fs, Path filename) {
+ // Vortex files are not splittable.
+ return false;
+ }
+}
diff --git
a/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/realtime/HoodieVortexRealtimeInputFormat.java
b/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/realtime/HoodieVortexRealtimeInputFormat.java
new file mode 100644
index 000000000000..789f8a1eacc5
--- /dev/null
+++
b/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/realtime/HoodieVortexRealtimeInputFormat.java
@@ -0,0 +1,64 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.hadoop.realtime;
+
+import org.apache.hudi.hadoop.UseFileSplitsFromInputFormat;
+import org.apache.hudi.hadoop.UseRecordReaderFromInputFormat;
+
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.io.ArrayWritable;
+import org.apache.hadoop.io.NullWritable;
+import org.apache.hadoop.mapred.InputSplit;
+import org.apache.hadoop.mapred.JobConf;
+import org.apache.hadoop.mapred.RecordReader;
+import org.apache.hadoop.mapred.Reporter;
+
+import java.io.IOException;
+
+/**
+ * HoodieRealtimeInputFormat for HUDI datasets which store data in Vortex base
file format.
+ * <p>
+ * This class is required for catalog/metastore registration during CREATE
TABLE operations
+ * for MOR tables with Vortex base file format.
+ * <p>
+ * TODO(#18623): Vortex reading through Hive InputFormat is not yet supported.
When support is
+ * added, this should route through {@code
HoodieFileGroupReaderBasedRecordReader} to get
+ * unified MOR log merging, schema evolution, and bootstrap support.
+ *
+ * @see <a href="https://github.com/apache/hudi/issues/18623";>#18623</a>
+ */
+@UseRecordReaderFromInputFormat
+@UseFileSplitsFromInputFormat
+public class HoodieVortexRealtimeInputFormat extends
HoodieMergeOnReadTableInputFormat {
+
+ @Override
+ public RecordReader<NullWritable, ArrayWritable> getRecordReader(final
InputSplit split, final JobConf jobConf,
+ final Reporter reporter) throws IOException {
+ throw new UnsupportedOperationException(
+ "Vortex reading through Hive InputFormat is not yet supported. "
+ + "Use the Spark datasource path (spark.read.format(\"hudi\")) to
read Vortex tables.");
+ }
+
+ @Override
+ protected boolean isSplitable(FileSystem fs, Path filename) {
+ // Vortex files are not splittable.
+ return false;
+ }
+}
diff --git
a/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/utils/HoodieInputFormatUtils.java
b/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/utils/HoodieInputFormatUtils.java
index 1d00061ca94f..44f73d5a6b37 100644
---
a/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/utils/HoodieInputFormatUtils.java
+++
b/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/utils/HoodieInputFormatUtils.java
@@ -43,6 +43,7 @@ import org.apache.hudi.hadoop.FileStatusWithBootstrapBaseFile;
import org.apache.hudi.hadoop.HoodieHFileInputFormat;
import org.apache.hudi.hadoop.HoodieLanceInputFormat;
import org.apache.hudi.hadoop.HoodieParquetInputFormat;
+import org.apache.hudi.hadoop.HoodieVortexInputFormat;
import org.apache.hudi.hadoop.LocatedFileStatusWithBootstrapBaseFile;
import org.apache.hudi.hadoop.fs.HadoopFSUtils;
import org.apache.hudi.hadoop.realtime.HoodieHFileRealtimeInputFormat;
@@ -50,6 +51,7 @@ import
org.apache.hudi.hadoop.realtime.HoodieLanceRealtimeInputFormat;
import org.apache.hudi.hadoop.realtime.HoodieParquetRealtimeInputFormat;
import org.apache.hudi.hadoop.realtime.HoodieRealtimeFileSplit;
import org.apache.hudi.hadoop.realtime.HoodieRealtimePath;
+import org.apache.hudi.hadoop.realtime.HoodieVortexRealtimeInputFormat;
import org.apache.hudi.hadoop.realtime.RealtimeSplit;
import org.apache.hudi.storage.HoodieStorage;
import org.apache.hudi.storage.StoragePath;
@@ -135,6 +137,16 @@ public class HoodieInputFormatUtils {
inputFormat.setConf(conf);
return inputFormat;
}
+ case VORTEX:
+ if (realtime) {
+ HoodieVortexRealtimeInputFormat inputFormat = new
HoodieVortexRealtimeInputFormat();
+ inputFormat.setConf(conf);
+ return inputFormat;
+ } else {
+ HoodieVortexInputFormat inputFormat = new HoodieVortexInputFormat();
+ inputFormat.setConf(conf);
+ return inputFormat;
+ }
default:
throw new HoodieIOException("Hoodie InputFormat not implemented for
base file format " + baseFileFormat);
}
@@ -164,6 +176,12 @@ public class HoodieInputFormatUtils {
} else {
return HoodieLanceInputFormat.class.getName();
}
+ case VORTEX:
+ if (realtime) {
+ return HoodieVortexRealtimeInputFormat.class.getName();
+ } else {
+ return HoodieVortexInputFormat.class.getName();
+ }
case ORC:
return OrcInputFormat.class.getName();
default:
@@ -176,6 +194,7 @@ public class HoodieInputFormatUtils {
case PARQUET:
case HFILE:
case LANCE:
+ case VORTEX:
return MapredParquetOutputFormat.class.getName();
case ORC:
return OrcOutputFormat.class.getName();
@@ -189,6 +208,7 @@ public class HoodieInputFormatUtils {
case PARQUET:
case HFILE:
case LANCE:
+ case VORTEX:
return ParquetHiveSerDe.class.getName();
case ORC:
return OrcSerde.class.getName();
@@ -208,6 +228,9 @@ public class HoodieInputFormatUtils {
if (extension.equals(HoodieFileFormat.LANCE.getFileExtension())) {
return getInputFormat(HoodieFileFormat.LANCE, realtime, conf);
}
+ if (extension.equals(HoodieFileFormat.VORTEX.getFileExtension())) {
+ return getInputFormat(HoodieFileFormat.VORTEX, realtime, conf);
+ }
// now we support read log file, try to find log file
if (HadoopFSUtils.isLogFile(new Path(path)) && realtime) {
return getInputFormat(HoodieFileFormat.PARQUET, realtime, conf);
diff --git a/hudi-spark-datasource/hudi-spark-common/pom.xml
b/hudi-spark-datasource/hudi-spark-common/pom.xml
index 355373334c19..1dca8ee367a8 100644
--- a/hudi-spark-datasource/hudi-spark-common/pom.xml
+++ b/hudi-spark-datasource/hudi-spark-common/pom.xml
@@ -264,4 +264,48 @@
</dependencies>
+ <profiles>
+ <!-- Vortex requires Java 17 (vortex-jni/vortex-spark ship Java 17
bytecode). Compile the Vortex
+ datasource reader under src/main/scala-vortex and pull the
vortex-spark connector only on
+ JDK 17; Java 11 builds skip it. Only the Spark 4.x adapters (Java 17)
wire it in; the Spark
+ 3.x adapters return None. -->
+ <profile>
+ <id>vortex</id>
+ <activation>
+ <jdk>[17,)</jdk>
+ </activation>
+ <dependencies>
+ <dependency>
+ <groupId>dev.vortex</groupId>
+ <artifactId>${vortex.spark.artifact}</artifactId>
+ </dependency>
+ </dependencies>
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.codehaus.mojo</groupId>
+ <artifactId>build-helper-maven-plugin</artifactId>
+ <executions>
+ <execution>
+ <id>add-vortex-source</id>
+ <phase>generate-sources</phase>
+ <goals>
+ <goal>add-source</goal>
+ </goals>
+ <configuration>
+ <!-- Vortex is only supported on Spark 3.5+/4.x; skip the
source on Spark 3.3/3.4
+ (vortex.skip.tests=true) where its Spark 3.5+ APIs are
unavailable. -->
+ <skipAddSource>${vortex.skip.tests}</skipAddSource>
+ <sources>
+ <source>src/main/scala-vortex</source>
+ </sources>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
+ </plugins>
+ </build>
+ </profile>
+ </profiles>
+
</project>
diff --git
a/hudi-spark-datasource/hudi-spark-common/src/main/scala-vortex/org/apache/spark/sql/execution/datasources/vortex/SparkVortexReaderBase.scala
b/hudi-spark-datasource/hudi-spark-common/src/main/scala-vortex/org/apache/spark/sql/execution/datasources/vortex/SparkVortexReaderBase.scala
new file mode 100644
index 000000000000..946638d41f49
--- /dev/null
+++
b/hudi-spark-datasource/hudi-spark-common/src/main/scala-vortex/org/apache/spark/sql/execution/datasources/vortex/SparkVortexReaderBase.scala
@@ -0,0 +1,203 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.spark.sql.execution.datasources.vortex
+
+import org.apache.hudi.SparkAdapterSupport.sparkAdapter
+import org.apache.hudi.common.config.HoodieStorageConfig
+import org.apache.hudi.common.schema.internal.InternalSchema
+import org.apache.hudi.common.util
+import org.apache.hudi.common.util.collection.ClosableIterator
+import org.apache.hudi.io.memory.HoodieArrowAllocator
+import org.apache.hudi.io.storage.VortexRecordIterator
+import org.apache.hudi.storage.StorageConfiguration
+
+import dev.vortex.api.{DataSource, ScanOptions, Session}
+import org.apache.hadoop.conf.Configuration
+import org.apache.parquet.schema.MessageType
+import org.apache.spark.TaskContext
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions.{AttributeReference,
JoinedRow, UnsafeProjection, UnsafeRow}
+import
org.apache.spark.sql.catalyst.expressions.codegen.GenerateUnsafeProjection
+import org.apache.spark.sql.execution.datasources.{PartitionedFile,
SparkColumnarFileReader, SparkSchemaTransformUtils}
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.sources.Filter
+import org.apache.spark.sql.types.StructType
+import org.apache.spark.sql.util.ArrowUtils
+
+import java.io.IOException
+import java.net.URI
+
+import scala.collection.JavaConverters._
+
+/**
+ * Reader for Vortex files in the Spark datasource query path (vortex-jni
0.76.0). Opens the file
+ * via `Session`/`DataSource`, scans it (reading all columns), and converts
Arrow batches to Spark
+ * rows through [[VortexRecordIterator]], which projects the requested columns
by name.
+ *
+ * Mirrors
[[org.apache.spark.sql.execution.datasources.lance.SparkLanceReaderBase]] for
the
+ * initial Vortex scope: standard Hudi data types, no filter pushdown/data
skipping, and no
+ * BLOB/VECTOR handling.
+ *
+ * @param enableVectorizedReader whether to use vectorized reading (currently
always true for Vortex)
+ */
+class SparkVortexReaderBase(enableVectorizedReader: Boolean) extends
SparkColumnarFileReader {
+
+ /**
+ * Read a Vortex file with schema projection and partition column support.
+ *
+ * @param file Vortex file to read
+ * @param requiredSchema desired output schema of the data (columns to
read)
+ * @param partitionSchema schema of the partition columns. Partition
values will be appended to the end of every row
+ * @param internalSchemaOpt option of internal schema for schema.on.read
(not currently used for Vortex)
+ * @param filters filters for data skipping. Not guaranteed to be
used; the spark plan will also apply the filters.
+ * @param storageConf the hadoop conf
+ * @param tableSchemaOpt option of table schema for timestamp precision
conversion fix.
+ * @return iterator of rows read from the file output type says
[[InternalRow]] but could be [[org.apache.spark.sql.vectorized.ColumnarBatch]]
+ */
+ override def read(file: PartitionedFile,
+ requiredSchema: StructType,
+ partitionSchema: StructType,
+ internalSchemaOpt: util.Option[InternalSchema],
+ filters: scala.Seq[Filter],
+ storageConf: StorageConfiguration[Configuration],
+ tableSchemaOpt: util.Option[MessageType] =
util.Option.empty()): Iterator[InternalRow] = {
+
+ val uri =
toVortexUri(sparkAdapter.getSparkPartitionedFileUtils.getPathFromPartitionedFile(file).toUri)
+
+ if (requiredSchema.isEmpty && partitionSchema.isEmpty) {
+ // No columns requested - return empty iterator
+ Iterator.empty
+ } else {
+
+ val dataAllocatorSize = storageConf.unwrap().getLong(
+ HoodieStorageConfig.VORTEX_READ_ALLOCATOR_SIZE_BYTES.key(),
+
HoodieStorageConfig.VORTEX_READ_ALLOCATOR_SIZE_BYTES.defaultValue().toLong)
+ val allocator = HoodieArrowAllocator.newChildAllocator(
+ getClass.getSimpleName + "-data-" + uri, dataAllocatorSize)
+
+ var vortexIterator: ClosableIterator[UnsafeRow] = null
+ var allocatorClosed = false
+
+ try {
+ // Session/DataSource are not AutoCloseable in vortex-jni 0.76.0
(native cleanup is implicit).
+ val session = Session.create()
+ val dataSource = DataSource.open(session, uri)
+
+ // Convert the Vortex file schema to a Spark schema, then reconcile it
against the requested
+ // schema for schema evolution (implicit type changes, columns not
present in the file).
+ val fileSchema =
ArrowUtils.fromArrowSchema(dataSource.arrowSchema(allocator))
+ val (implicitTypeChangeInfo, sparkRequestSchema) =
+ SparkSchemaTransformUtils.buildImplicitSchemaChangeInfo(fileSchema,
requiredSchema)
+
+ // Only read columns that actually exist in the file; missing columns
are null-padded below.
+ val requestSchema =
+
SparkSchemaTransformUtils.filterSchemaByFileSchema(sparkRequestSchema,
fileSchema)
+
+ val baseIter: Iterator[InternalRow] = if (requestSchema.isEmpty) {
+ // No data columns to read from the file (only partition columns
requested, or all
+ // requested columns are new via schema evolution). Emit one empty
row per record so the
+ // null-padding and partition projections can fill in the values.
+ val numRows = dataSource.rowCount().asOptional().orElse(0L)
+ allocator.close()
+ allocatorClosed = true
+ val emptyRow = InternalRow.empty
+ // Emit numRows empty rows without narrowing the long count to int
(a >2^31-row file
+ // would otherwise overflow and silently yield zero rows).
+ new Iterator[InternalRow] {
+ private var remaining = numRows
+ override def hasNext: Boolean = remaining > 0
+ override def next(): InternalRow = {
+ remaining -= 1
+ emptyRow
+ }
+ }
+ } else {
+ val scan = dataSource.scan(ScanOptions.of())
+ vortexIterator = new VortexRecordIterator(allocator, session,
dataSource, scan, requestSchema, uri)
+
+ // Register cleanup listener
+ Option(TaskContext.get()).foreach { ctx =>
+ ctx.addTaskCompletionListener[Unit](_ => vortexIterator.close())
+ }
+
+ vortexIterator.asScala
+ }
+
+ // Create the following projections for schema evolution:
+ // 1. Padding projection: add NULL for missing columns
+ // 2. Casting projection: handle implicit type conversions
+ val schemaUtils = sparkAdapter.getSchemaUtils
+ val paddingProj =
SparkSchemaTransformUtils.generateNullPaddingProjection(requestSchema,
requiredSchema)
+ val castProj = SparkSchemaTransformUtils.generateUnsafeProjection(
+ schemaUtils.toAttributes(requiredSchema),
+ Some(SQLConf.get.sessionLocalTimeZone),
+ implicitTypeChangeInfo,
+ requiredSchema,
+ new StructType(),
+ schemaUtils
+ )
+
+ // Unify projections by applying padding and then casting for each row
+ val projection: UnsafeProjection = new UnsafeProjection {
+ def apply(row: InternalRow): UnsafeRow =
+ castProj(paddingProj(row))
+ }
+ val projectedIter = baseIter.map(projection.apply)
+
+ // Handle partition columns
+ if (partitionSchema.length == 0) {
+ // No partition columns - return rows directly
+ projectedIter
+ } else {
+ // Create UnsafeProjection to convert JoinedRow to UnsafeRow
+ val fullSchema = (requiredSchema.fields ++
partitionSchema.fields).map(f =>
+ AttributeReference(f.name, f.dataType, f.nullable, f.metadata)())
+ val unsafeProjection = GenerateUnsafeProjection.generate(fullSchema,
fullSchema)
+
+ // Append partition values to each row using JoinedRow, then convert
to UnsafeRow
+ val joinedRow = new JoinedRow()
+ projectedIter.map(row => unsafeProjection(joinedRow(row,
file.partitionValues)))
+ }
+ } catch {
+ case e: Exception =>
+ if (vortexIterator != null) {
+ vortexIterator.close() // Closes the iterator, which owns the
allocator lifecycle
+ } else if (!allocatorClosed) {
+ allocator.close()
+ }
+ throw new IOException(s"Failed to read Vortex file: $uri", e)
+ }
+ }
+ }
+
+ /**
+ * Vortex requires a well-formed URL. Remote-store paths already carry a
scheme
+ * (s3://, hdfs://, ...); local paths may not, so qualify them as `file://`
URIs. Takes an
+ * already-parsed URI (from the storage path) rather than re-parsing a
display string, so paths
+ * with spaces or other characters needing URI encoding are handled
correctly.
+ */
+ private def toVortexUri(parsed: URI): String = {
+ if (parsed.getScheme == null) {
+ new java.io.File(parsed.getPath).toURI.toString
+ } else {
+ parsed.toString
+ }
+ }
+}
diff --git
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieMergeOnReadRDDV2.scala
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieMergeOnReadRDDV2.scala
index e02931b778ef..bb321616a1eb 100644
---
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieMergeOnReadRDDV2.scala
+++
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieMergeOnReadRDDV2.scala
@@ -117,7 +117,8 @@ class HoodieMergeOnReadRDDV2(@transient sc: SparkContext,
val parquetReader = sparkAdapter.createParquetFileReader(vectorized =
false, sqlConf, updatedOptions, config)
val orcReader = sparkAdapter.createOrcFileReader(vectorized = false,
sqlConf, updatedOptions, config, tableSchema.structTypeSchema)
val lanceReader = sparkAdapter.createLanceFileReader(vectorized =
false, sqlConf, updatedOptions, config).orNull
- val multiReader = new MultipleColumnarFileFormatReader(parquetReader,
orcReader, lanceReader)
+ val vortexReader = sparkAdapter.createVortexFileReader(vectorized =
false, sqlConf, updatedOptions, config).orNull
+ val multiReader = new MultipleColumnarFileFormatReader(parquetReader,
orcReader, lanceReader, vortexReader)
sc.broadcast(multiReader)
} else if (metaClient.getTableConfig.getBaseFileFormat ==
HoodieFileFormat.PARQUET) {
sc.broadcast(sparkAdapter.createParquetFileReader(vectorized = false,
sqlConf, updatedOptions, config))
@@ -125,6 +126,8 @@ class HoodieMergeOnReadRDDV2(@transient sc: SparkContext,
sc.broadcast(sparkAdapter.createOrcFileReader(vectorized = false,
sqlConf, updatedOptions, config, tableSchema.structTypeSchema))
} else if (metaClient.getTableConfig.getBaseFileFormat ==
HoodieFileFormat.LANCE) {
sc.broadcast(sparkAdapter.createLanceFileReader(vectorized = false,
sqlConf, updatedOptions, config).orNull)
+ } else if (metaClient.getTableConfig.getBaseFileFormat ==
HoodieFileFormat.VORTEX) {
+ sc.broadcast(sparkAdapter.createVortexFileReader(vectorized = false,
sqlConf, updatedOptions, config).orNull)
} else {
throw new IllegalArgumentException(s"Unsupported base file format:
${metaClient.getTableConfig.getBaseFileFormat}")
}
diff --git
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/HoodieFileGroupReaderBasedFileFormat.scala
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/HoodieFileGroupReaderBasedFileFormat.scala
index 429b769d5b7a..4b4918ad2a7c 100644
---
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/HoodieFileGroupReaderBasedFileFormat.scala
+++
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/HoodieFileGroupReaderBasedFileFormat.scala
@@ -363,13 +363,16 @@ class HoodieFileGroupReaderBasedFileFormat(tablePath:
String,
val parquetReader =
sparkAdapter.createParquetFileReader(enableVectorizedRead,
spark.sessionState.conf, options, configuration)
val orcReader = sparkAdapter.createOrcFileReader(enableVectorizedRead,
spark.sessionState.conf, options, configuration, dataSchema)
val lanceReader =
sparkAdapter.createLanceFileReader(enableVectorizedRead,
spark.sessionState.conf, options, configuration).orNull
- new MultipleColumnarFileFormatReader(parquetReader, orcReader,
lanceReader)
+ val vortexReader =
sparkAdapter.createVortexFileReader(enableVectorizedRead,
spark.sessionState.conf, options, configuration).orNull
+ new MultipleColumnarFileFormatReader(parquetReader, orcReader,
lanceReader, vortexReader)
} else if (hoodieFileFormat == HoodieFileFormat.PARQUET) {
sparkAdapter.createParquetFileReader(enableVectorizedRead,
spark.sessionState.conf, options, configuration)
} else if (hoodieFileFormat == HoodieFileFormat.ORC) {
sparkAdapter.createOrcFileReader(enableVectorizedRead,
spark.sessionState.conf, options, configuration, dataSchema)
} else if (hoodieFileFormat == HoodieFileFormat.LANCE) {
sparkAdapter.createLanceFileReader(enableVectorizedRead,
spark.sessionState.conf, options, configuration).orNull
+ } else if (hoodieFileFormat == HoodieFileFormat.VORTEX) {
+ sparkAdapter.createVortexFileReader(enableVectorizedRead,
spark.sessionState.conf, options, configuration).orNull
} else {
throw new HoodieNotSupportedException("Unsupported file format: " +
hoodieFileFormat)
}
diff --git a/hudi-spark-datasource/hudi-spark/pom.xml
b/hudi-spark-datasource/hudi-spark/pom.xml
index efc0f3391523..bf7e2dbc5ce3 100644
--- a/hudi-spark-datasource/hudi-spark/pom.xml
+++ b/hudi-spark-datasource/hudi-spark/pom.xml
@@ -498,4 +498,47 @@
<scope>test</scope>
</dependency>
</dependencies>
+
+ <profiles>
+ <!-- Vortex requires Java 17 (vortex-jni/vortex-spark ship Java 17
bytecode). The Vortex
+ round-trip test compiles/runs only under JDK 17; Java 11 builds skip
it. -->
+ <profile>
+ <id>vortex</id>
+ <activation>
+ <jdk>[17,)</jdk>
+ </activation>
+ <dependencies>
+ <dependency>
+ <groupId>dev.vortex</groupId>
+ <artifactId>${vortex.spark.artifact}</artifactId>
+ <scope>test</scope>
+ </dependency>
+ </dependencies>
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.codehaus.mojo</groupId>
+ <artifactId>build-helper-maven-plugin</artifactId>
+ <executions>
+ <execution>
+ <id>add-vortex-test-source</id>
+ <phase>generate-test-sources</phase>
+ <goals>
+ <goal>add-test-source</goal>
+ </goals>
+ <configuration>
+ <!-- Vortex is only supported on Spark 3.5+/4.x; skip the
test on Spark 3.3/3.4
+ (vortex.skip.tests=true) where its Spark 3.5+ APIs are
unavailable. -->
+ <skipAddTestSource>${vortex.skip.tests}</skipAddTestSource>
+ <sources>
+ <source>src/test/java-vortex</source>
+ </sources>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
+ </plugins>
+ </build>
+ </profile>
+ </profiles>
</project>
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/java-vortex/org/apache/hudi/io/storage/TestVortexReaderWriterRoundTrip.java
b/hudi-spark-datasource/hudi-spark/src/test/java-vortex/org/apache/hudi/io/storage/TestVortexReaderWriterRoundTrip.java
new file mode 100644
index 000000000000..8a8764f07d50
--- /dev/null
+++
b/hudi-spark-datasource/hudi-spark/src/test/java-vortex/org/apache/hudi/io/storage/TestVortexReaderWriterRoundTrip.java
@@ -0,0 +1,141 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.io.storage;
+
+import org.apache.hudi.HoodieSchemaConversionUtils;
+import org.apache.hudi.client.SparkTaskContextSupplier;
+import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.common.schema.HoodieSchema;
+import org.apache.hudi.common.testutils.HoodieTestUtils;
+import org.apache.hudi.common.util.collection.ClosableIterator;
+import org.apache.hudi.storage.HoodieStorage;
+import org.apache.hudi.storage.StoragePath;
+
+import org.apache.spark.sql.catalyst.InternalRow;
+import org.apache.spark.sql.catalyst.expressions.GenericInternalRow;
+import org.apache.spark.sql.types.DataTypes;
+import org.apache.spark.sql.types.StructType;
+import org.apache.spark.unsafe.types.UTF8String;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.condition.DisabledIfSystemProperty;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.io.File;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * End-to-end round-trip test for the Vortex file writer and reader against
the real vortex-jni
+ * API. Writes InternalRows through {@link HoodieSparkVortexWriter} and reads
them back through
+ * {@link HoodieSparkVortexReader}, spanning multiple flushed batches.
+ */
+@DisabledIfSystemProperty(named = "vortex.skip.tests", matches = "true")
+public class TestVortexReaderWriterRoundTrip {
+
+ @TempDir
+ File tempDir;
+
+ @Test
+ public void testWriteReadRoundTrip() throws Exception {
+ SparkTaskContextSupplier taskContextSupplier = new
SparkTaskContextSupplier();
+
+ StructType schema = new StructType()
+ .add("id", DataTypes.LongType, false)
+ .add("name", DataTypes.StringType, false)
+ .add("score", DataTypes.DoubleType, false);
+
+ StoragePath path = new StoragePath(tempDir.getAbsolutePath() +
"/roundtrip.vortex");
+
+ int numRows = 2500; // > default batch size (1000) to force multiple
flushed batches
+ try (HoodieStorage storage =
HoodieTestUtils.getStorage(tempDir.getAbsolutePath())) {
+ try (HoodieSparkVortexWriter writer = HoodieSparkVortexWriter.builder()
+ .file(path)
+ .sparkSchema(schema)
+ .instantTime("20251201120000000")
+ .taskContextSupplier(taskContextSupplier)
+ .storage(storage)
+ .populateMetaFields(false)
+ .build()) {
+ for (int i = 0; i < numRows; i++) {
+ InternalRow row = new GenericInternalRow(new Object[] {
+ (long) (100 + i), UTF8String.fromString("name" + i), (double) i
+ 0.5});
+ writer.writeRow(row);
+ }
+ }
+
+ assertTrue(storage.exists(path), "Vortex file should exist");
+
+ try (HoodieSparkVortexReader reader = new HoodieSparkVortexReader(path))
{
+ assertEquals(numRows, reader.getTotalRecords(), "row count");
+
+ HoodieSchema readSchema = reader.getSchema();
+ assertEquals(3, readSchema.getFields().size(), "schema field count");
+
+ int count = 0;
+ try (ClosableIterator<HoodieRecord<InternalRow>> it =
reader.getRecordIterator(readSchema)) {
+ while (it.hasNext()) {
+ InternalRow row = it.next().getData();
+ assertEquals(100L + count, row.getLong(0), "id at row " + count);
+ assertEquals("name" + count, row.getUTF8String(1).toString(),
"name at row " + count);
+ assertEquals((double) count + 0.5, row.getDouble(2), 1e-9, "score
at row " + count);
+ count++;
+ }
+ }
+ assertEquals(numRows, count, "rows read back");
+ }
+ }
+ }
+
+ @Test
+ public void testProjectedRead() throws Exception {
+ SparkTaskContextSupplier taskContextSupplier = new
SparkTaskContextSupplier();
+
+ StructType schema = new StructType()
+ .add("id", DataTypes.LongType, false)
+ .add("name", DataTypes.StringType, false);
+
+ StoragePath path = new StoragePath(tempDir.getAbsolutePath() +
"/projected.vortex");
+ try (HoodieStorage storage =
HoodieTestUtils.getStorage(tempDir.getAbsolutePath())) {
+ try (HoodieSparkVortexWriter writer = HoodieSparkVortexWriter.builder()
+ .file(path).sparkSchema(schema).instantTime("20251201120000000")
+
.taskContextSupplier(taskContextSupplier).storage(storage).populateMetaFields(false).build())
{
+ for (int i = 0; i < 3; i++) {
+ writer.writeRow(new GenericInternalRow(new Object[] {(long) i,
UTF8String.fromString("n" + i)}));
+ }
+ }
+
+ // Project only the non-leading "name" column so the assertion fails if
the reader returns the
+ // full row instead of narrowing to the requested schema (projecting the
leading "id" column
+ // would read back at index 0 either way and could not detect that
regression).
+ HoodieSchema projected =
HoodieSchemaConversionUtils.convertStructTypeToHoodieSchema(
+ new StructType().add("name", DataTypes.StringType, false), "record",
"", false);
+ try (HoodieSparkVortexReader reader = new HoodieSparkVortexReader(path);
+ ClosableIterator<HoodieRecord<InternalRow>> it =
reader.getRecordIterator(projected)) {
+ int count = 0;
+ while (it.hasNext()) {
+ InternalRow row = it.next().getData();
+ assertEquals("n" + count, row.getUTF8String(0).toString(),
"projected name at row " + count);
+ count++;
+ }
+ assertEquals(3, count, "projected rows");
+ }
+ }
+ }
+}
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/common/table/read/TestHoodieFileGroupReaderOnSpark.scala
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/common/table/read/TestHoodieFileGroupReaderOnSpark.scala
index 8c4577192641..243ea0ade045 100644
---
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/common/table/read/TestHoodieFileGroupReaderOnSpark.scala
+++
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/common/table/read/TestHoodieFileGroupReaderOnSpark.scala
@@ -116,7 +116,8 @@ class TestHoodieFileGroupReaderOnSpark extends
TestHoodieFileGroupReaderBase[Int
val dataSchema =
HoodieSchemaConversionUtils.convertHoodieSchemaToStructType(schema)
val orcReader = sparkAdapter.createOrcFileReader(vectorized = false,
spark.sessionState.conf, Map.empty,
storageConf.unwrapAs(classOf[Configuration]), dataSchema)
val lanceReader = sparkAdapter.createLanceFileReader(vectorized = false,
spark.sessionState.conf, Map.empty,
storageConf.unwrapAs(classOf[Configuration])).orNull
- val multiFormatReader = new
MultipleColumnarFileFormatReader(parquetReader, orcReader, lanceReader)
+ val vortexReader = sparkAdapter.createVortexFileReader(vectorized = false,
spark.sessionState.conf, Map.empty,
storageConf.unwrapAs(classOf[Configuration])).orNull
+ val multiFormatReader = new
MultipleColumnarFileFormatReader(parquetReader, orcReader, lanceReader,
vortexReader)
new SparkFileFormatInternalRowReaderContext(multiFormatReader, Seq.empty,
Seq.empty, getStorageConf, metaClient.getTableConfig)
}
diff --git
a/hudi-spark-datasource/hudi-spark3.3.x/src/main/scala/org/apache/spark/sql/adapter/Spark3_3Adapter.scala
b/hudi-spark-datasource/hudi-spark3.3.x/src/main/scala/org/apache/spark/sql/adapter/Spark3_3Adapter.scala
index 25553f8ebdd7..b406668b908b 100644
---
a/hudi-spark-datasource/hudi-spark3.3.x/src/main/scala/org/apache/spark/sql/adapter/Spark3_3Adapter.scala
+++
b/hudi-spark-datasource/hudi-spark3.3.x/src/main/scala/org/apache/spark/sql/adapter/Spark3_3Adapter.scala
@@ -165,6 +165,13 @@ class Spark3_3Adapter extends BaseSpark3Adapter {
None
}
+ override def createVortexFileReader(vectorized: Boolean,
+ sqlConf: SQLConf,
+ options: Map[String, String],
+ hadoopConf: Configuration):
Option[SparkColumnarFileReader] = {
+ None
+ }
+
override def stopSparkContext(jssc: JavaSparkContext, exitCode: Int): Unit =
{
jssc.stop()
}
diff --git
a/hudi-spark-datasource/hudi-spark3.4.x/src/main/scala/org/apache/spark/sql/adapter/Spark3_4Adapter.scala
b/hudi-spark-datasource/hudi-spark3.4.x/src/main/scala/org/apache/spark/sql/adapter/Spark3_4Adapter.scala
index 3d62ba1edc0c..93841ecf5b4c 100644
---
a/hudi-spark-datasource/hudi-spark3.4.x/src/main/scala/org/apache/spark/sql/adapter/Spark3_4Adapter.scala
+++
b/hudi-spark-datasource/hudi-spark3.4.x/src/main/scala/org/apache/spark/sql/adapter/Spark3_4Adapter.scala
@@ -166,6 +166,14 @@ class Spark3_4Adapter extends BaseSpark3Adapter {
Some(new SparkLanceReaderBase(vectorized))
}
+ override def createVortexFileReader(vectorized: Boolean,
+ sqlConf: SQLConf,
+ options: Map[String, String],
+ hadoopConf: Configuration):
Option[SparkColumnarFileReader] = {
+ // Vortex Spark connector artifact is not available for Spark 3.4.
+ None
+ }
+
override def stopSparkContext(jssc: JavaSparkContext, exitCode: Int): Unit =
{
jssc.sc.stop(exitCode)
}
diff --git
a/hudi-spark-datasource/hudi-spark3.5.x/src/main/scala/org/apache/spark/sql/adapter/Spark3_5Adapter.scala
b/hudi-spark-datasource/hudi-spark3.5.x/src/main/scala/org/apache/spark/sql/adapter/Spark3_5Adapter.scala
index 3fd01ae7085b..5cad378dc736 100644
---
a/hudi-spark-datasource/hudi-spark3.5.x/src/main/scala/org/apache/spark/sql/adapter/Spark3_5Adapter.scala
+++
b/hudi-spark-datasource/hudi-spark3.5.x/src/main/scala/org/apache/spark/sql/adapter/Spark3_5Adapter.scala
@@ -181,6 +181,15 @@ class Spark3_5Adapter extends BaseSpark3Adapter {
Some(new SparkLanceReaderBase(vectorized))
}
+ override def createVortexFileReader(vectorized: Boolean,
+ sqlConf: SQLConf,
+ options: Map[String, String],
+ hadoopConf: Configuration):
Option[SparkColumnarFileReader] = {
+ // Vortex requires Java 17 (vortex-jni ships Java 17 bytecode) and is
wired only through the
+ // Spark 4.x adapters. Spark 3.5 runs on Java 11, where
SparkVortexReaderBase is not compiled.
+ None
+ }
+
override def stopSparkContext(jssc: JavaSparkContext, exitCode: Int): Unit =
{
jssc.sc.stop(exitCode)
}
diff --git
a/hudi-spark-datasource/hudi-spark4.0.x/src/main/scala/org/apache/spark/sql/adapter/Spark4_0Adapter.scala
b/hudi-spark-datasource/hudi-spark4.0.x/src/main/scala/org/apache/spark/sql/adapter/Spark4_0Adapter.scala
index 519ec97cd771..f622b266db9f 100644
---
a/hudi-spark-datasource/hudi-spark4.0.x/src/main/scala/org/apache/spark/sql/adapter/Spark4_0Adapter.scala
+++
b/hudi-spark-datasource/hudi-spark4.0.x/src/main/scala/org/apache/spark/sql/adapter/Spark4_0Adapter.scala
@@ -45,6 +45,7 @@ import org.apache.spark.sql.execution.datasources._
import org.apache.spark.sql.execution.datasources.lance.SparkLanceReaderBase
import
org.apache.spark.sql.execution.datasources.parquet.{HoodieParquetReadSupport,
ParquetFileFormat, Spark40HoodieParquetReadSupport,
Spark40LegacyHoodieParquetFileFormat, Spark40ParquetReader}
import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Relation
+import org.apache.spark.sql.execution.datasources.vortex.SparkVortexReaderBase
import org.apache.spark.sql.execution.streaming.MemoryStream
import org.apache.spark.sql.hudi.HoodieMemoryStream
import org.apache.spark.sql.hudi.analysis.TableValuedFunctions
@@ -205,6 +206,13 @@ class Spark4_0Adapter extends BaseSpark4Adapter {
Some(new SparkLanceReaderBase(vectorized))
}
+ override def createVortexFileReader(vectorized: Boolean,
+ sqlConf: SQLConf,
+ options: Map[String, String],
+ hadoopConf: Configuration):
Option[SparkColumnarFileReader] = {
+ Some(new SparkVortexReaderBase(vectorized))
+ }
+
override def stopSparkContext(jssc: JavaSparkContext, exitCode: Int): Unit =
{
jssc.sc.stop(exitCode)
}
diff --git
a/hudi-spark-datasource/hudi-spark4.1.x/src/main/scala/org/apache/spark/sql/adapter/Spark4_1Adapter.scala
b/hudi-spark-datasource/hudi-spark4.1.x/src/main/scala/org/apache/spark/sql/adapter/Spark4_1Adapter.scala
index f6a8e19b0e6c..363d0b64446a 100644
---
a/hudi-spark-datasource/hudi-spark4.1.x/src/main/scala/org/apache/spark/sql/adapter/Spark4_1Adapter.scala
+++
b/hudi-spark-datasource/hudi-spark4.1.x/src/main/scala/org/apache/spark/sql/adapter/Spark4_1Adapter.scala
@@ -45,6 +45,7 @@ import org.apache.spark.sql.execution.datasources._
import org.apache.spark.sql.execution.datasources.lance.SparkLanceReaderBase
import org.apache.spark.sql.execution.datasources.parquet.{ParquetFileFormat,
Spark41LegacyHoodieParquetFileFormat, Spark41ParquetReader}
import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Relation
+import org.apache.spark.sql.execution.datasources.vortex.SparkVortexReaderBase
import org.apache.spark.sql.execution.streaming.runtime.MemoryStream
import org.apache.spark.sql.hudi.{HoodieMemoryStream, SparkAdapter}
import org.apache.spark.sql.hudi.analysis.TableValuedFunctions
@@ -196,6 +197,13 @@ class Spark4_1Adapter extends BaseSpark4Adapter {
Some(new SparkLanceReaderBase(vectorized))
}
+ override def createVortexFileReader(vectorized: Boolean,
+ sqlConf: SQLConf,
+ options: Map[String, String],
+ hadoopConf: Configuration):
Option[SparkColumnarFileReader] = {
+ Some(new SparkVortexReaderBase(vectorized))
+ }
+
override def stopSparkContext(jssc: JavaSparkContext, exitCode: Int): Unit =
{
jssc.sc.stop(exitCode)
}
diff --git
a/hudi-spark-datasource/hudi-spark4.2.x/src/main/scala/org/apache/spark/sql/adapter/Spark4_2Adapter.scala
b/hudi-spark-datasource/hudi-spark4.2.x/src/main/scala/org/apache/spark/sql/adapter/Spark4_2Adapter.scala
index a0fc4e5ee843..b2790091b5fd 100644
---
a/hudi-spark-datasource/hudi-spark4.2.x/src/main/scala/org/apache/spark/sql/adapter/Spark4_2Adapter.scala
+++
b/hudi-spark-datasource/hudi-spark4.2.x/src/main/scala/org/apache/spark/sql/adapter/Spark4_2Adapter.scala
@@ -45,6 +45,7 @@ import org.apache.spark.sql.execution.datasources._
import org.apache.spark.sql.execution.datasources.lance.SparkLanceReaderBase
import org.apache.spark.sql.execution.datasources.parquet.{ParquetFileFormat,
Spark42LegacyHoodieParquetFileFormat, Spark42ParquetReader}
import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Relation
+import org.apache.spark.sql.execution.datasources.vortex.SparkVortexReaderBase
import org.apache.spark.sql.execution.streaming.runtime.MemoryStream
import org.apache.spark.sql.hudi.{HoodieMemoryStream, SparkAdapter}
import org.apache.spark.sql.hudi.analysis.TableValuedFunctions
@@ -196,6 +197,13 @@ class Spark4_2Adapter extends BaseSpark4Adapter {
Some(new SparkLanceReaderBase(vectorized))
}
+ override def createVortexFileReader(vectorized: Boolean,
+ sqlConf: SQLConf,
+ options: Map[String, String],
+ hadoopConf: Configuration):
Option[SparkColumnarFileReader] = {
+ Some(new SparkVortexReaderBase(vectorized))
+ }
+
override def stopSparkContext(jssc: JavaSparkContext, exitCode: Int): Unit =
{
jssc.sc.stop(exitCode)
}
diff --git a/pom.xml b/pom.xml
index 2523a3c7b5f6..4ae9a2309cf3 100644
--- a/pom.xml
+++ b/pom.xml
@@ -253,6 +253,9 @@
<lance.spark.connector.version>0.4.0</lance.spark.connector.version>
<lance.spark.artifact>lance-spark-3.5_${scala.binary.version}</lance.spark.artifact>
<lance.skip.tests>false</lance.skip.tests>
+ <vortex.version>0.76.0</vortex.version>
+
<vortex.spark.artifact>vortex-spark_${scala.binary.version}</vortex.spark.artifact>
+ <vortex.skip.tests>false</vortex.skip.tests>
<!-- The following properties are only used for Jacoco coverage report
aggregation -->
<copy.files>false</copy.files>
<copy.files.target.dir>${maven.multiModuleProjectDirectory}</copy.files.target.dir>
@@ -525,6 +528,7 @@
<systemPropertyVariables>
<log4j.configurationFile>${surefire-log4j.file}</log4j.configurationFile>
<lance.skip.tests>${lance.skip.tests}</lance.skip.tests>
+ <vortex.skip.tests>${vortex.skip.tests}</vortex.skip.tests>
</systemPropertyVariables>
<useSystemClassLoader>false</useSystemClassLoader>
<forkedProcessExitTimeoutInSeconds>30</forkedProcessExitTimeoutInSeconds>
@@ -569,6 +573,7 @@
<systemProperties>
<log4j.configurationFile>${surefire-log4j.file}</log4j.configurationFile>
<lance.skip.tests>${lance.skip.tests}</lance.skip.tests>
+ <vortex.skip.tests>${vortex.skip.tests}</vortex.skip.tests>
</systemProperties>
</configuration>
<executions>
@@ -996,6 +1001,24 @@
</exclusion>
</exclusions>
</dependency>
+ <!-- Vortex JNI -->
+ <dependency>
+ <groupId>dev.vortex</groupId>
+ <artifactId>vortex-jni</artifactId>
+ <version>${vortex.version}</version>
+ </dependency>
+ <!-- Vortex Spark connector -->
+ <dependency>
+ <groupId>dev.vortex</groupId>
+ <artifactId>${vortex.spark.artifact}</artifactId>
+ <version>${vortex.version}</version>
+ <exclusions>
+ <exclusion>
+ <groupId>dev.vortex</groupId>
+ <artifactId>vortex-jni</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
<!-- Apache Arrow -->
<dependency>
<groupId>org.apache.arrow</groupId>
@@ -2633,6 +2656,8 @@
<!-- Lance: Skip tests for Spark 3.3 due to lack of support -->
<lance.spark.artifact>lance-spark-base_${scala.binary.version}</lance.spark.artifact>
<lance.skip.tests>true</lance.skip.tests>
+ <!-- Vortex: Skip tests for Spark 3.3 due to lack of support -->
+ <vortex.skip.tests>true</vortex.skip.tests>
<!-- NOTE: Some Hudi modules require standalone Parquet/Orc/etc
file-format dependency (hudi-hive-sync,
hudi-hadoop-mr, for ex). Since these Hudi modules might be
used from w/in the execution engine(s)
bringing these file-formats as dependencies as well, we
need to make sure that versions are
@@ -2678,6 +2703,8 @@
<!-- Lance: Use Spark 3.4-specific artifact -->
<lance.spark.artifact>lance-spark-3.4_${scala.binary.version}</lance.spark.artifact>
<lance.skip.tests>true</lance.skip.tests>
+ <!-- Vortex: Skip tests for Spark 3.4 due to lack of support -->
+ <vortex.skip.tests>true</vortex.skip.tests>
<!-- NOTE: Some Hudi modules require standalone Parquet/Orc/etc
file-format dependency (hudi-hive-sync,
hudi-hadoop-mr, for ex). Since these Hudi modules might be
used from w/in the execution engine(s)
bringing these file-formats as dependencies as well, we
need to make sure that versions are
@@ -2733,6 +2760,9 @@
<!-- Lance: Use Spark 3.5-specific artifact -->
<lance.spark.artifact>lance-spark-3.5_${scala.binary.version}</lance.spark.artifact>
<lance.skip.tests>false</lance.skip.tests>
+ <!-- Vortex: Use Spark 3.5 artifact -->
+
<vortex.spark.artifact>vortex-spark_${scala.binary.version}</vortex.spark.artifact>
+ <vortex.skip.tests>false</vortex.skip.tests>
<!-- NOTE: Some Hudi modules require standalone Parquet/Orc/etc
file-format dependency (hudi-hive-sync,
hudi-hadoop-mr, for ex). Since these Hudi modules might be
used from w/in the execution engine(s)
bringing these file-formats as dependencies as well, we
need to make sure that versions are
@@ -2796,6 +2826,9 @@
<!-- Lance: Use Spark 4.0-specific artifact (Scala 2.13 only) -->
<lance.spark.artifact>lance-spark-4.0_2.13</lance.spark.artifact>
<lance.skip.tests>false</lance.skip.tests>
+ <!-- Vortex: Use Spark 4.0 artifact (Scala 2.13 only) -->
+ <vortex.spark.artifact>vortex-spark_2.13</vortex.spark.artifact>
+ <vortex.skip.tests>false</vortex.skip.tests>
<!-- NOTE: Some Hudi modules require standalone Parquet/Orc/etc
file-format dependency (hudi-hive-sync,
hudi-hadoop-mr, for ex). Since these Hudi modules might be
used from w/in the execution engine(s)
bringing these file-formats as dependencies as well, we
need to make sure that versions are
@@ -2859,6 +2892,9 @@
<hive.storage.version>2.8.1</hive.storage.version>
<lance.spark.artifact>lance-spark-4.1_2.13</lance.spark.artifact>
<lance.skip.tests>false</lance.skip.tests>
+ <!-- Vortex: Use Spark 4.1 artifact (Scala 2.13 only) -->
+ <vortex.spark.artifact>vortex-spark_2.13</vortex.spark.artifact>
+ <vortex.skip.tests>false</vortex.skip.tests>
<!-- NOTE: Some Hudi modules require standalone Parquet/Orc/etc
file-format dependency (hudi-hive-sync,
hudi-hadoop-mr, for ex). Since these Hudi modules might be
used from w/in the execution engine(s)
bringing these file-formats as dependencies as well, we
need to make sure that versions are
