This is an automated email from the ASF dual-hosted git repository.
vinothchandar 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 8e18999a4164 feat(ingest): ingest unstructured files into BLOB+VECTOR
Hudi tables via Hudi Streamer (#19278)
8e18999a4164 is described below
commit 8e18999a41645d4ae2fb240361615ae2430880c9
Author: vinoth chandar <[email protected]>
AuthorDate: Tue Jul 28 20:16:15 2026 +0530
feat(ingest): ingest unstructured files into BLOB+VECTOR Hudi tables via
Hudi Streamer (#19278)
* feat(ingest): add UnstructuredFileDFSSource for landing unstructured
files as BLOB tables
HoodieStreamer can now point at a DFS folder of arbitrary files (documents,
images, videos) and produce a queryable Hudi table using the 1.2 BLOB type
on plain Parquet base files:
- UnstructuredFileDFSSource (RowSource) reuses DFSPathSelector for
mtime-checkpointed incremental discovery; each sync ingests only new or
modified files, and keying on path with modification_time ordering makes
re-ingested files upsert in place.
- Blob placement is decided per record by a configurable size threshold
(default 1MiB): small files store bytes INLINE; larger files store an
OUT_OF_LINE reference to the original file in place (managed=false), so
large-file bytes never enter Spark rows -- memory and shuffle stay bounded
regardless of file sizes (source-limit already caps total inline bytes).
- Text extraction runs embedded in executors behind a pluggable
DocumentParser; the default TikaDocumentParser (Apache Tika, in-process,
no services) never throws: per-row parse_status/parse_error record
SUCCESS/TRUNCATED/EMPTY/SKIPPED/FAILED outcomes. Extracted text is chunked
(size/overlap configurable) into a nested chunks column for retrieval use.
- Only tika-core (~800KB) ships in the utilities bundles; parser modules for
PDF/Office/etc. are supplied at runtime (e.g. spark-submit packages
org.apache.tika:tika-parsers-standard-package), degrading gracefully to
EMPTY parses when absent.
Verified: unit tests for chunker boundaries, parser status mapping on
generated fixtures, source-level inline/out-of-line + checkpoint semantics,
BLOB logical type surviving StructType->Avro conversion, and a streamer E2E
(two syncs into a Parquet COW table: blob struct round-trip both placements,
upsert refresh on file change).
* feat(ingest): add EmbeddingTransformer populating a VECTOR column from an
embeddings API
Chains after UnstructuredFileDFSSource (or any source with a text column) to
keep a VECTOR(dimension) embedding column current with ingested data:
- Record-level batching inside each partition: up to batch.size (default
1024) records' texts per API request, then the next buffer. Executors stay
busy end to end and large request batches keep the request rate low
without any parallelism tuning knobs.
- Pluggable EmbeddingProvider; the default calls any OpenAI-compatible
/v1/embeddings endpoint (Ollama, TEI, vLLM, OpenAI, Voyage) with
exponential backoff honoring Retry-After on 429/5xx. API keys come from an
environment variable named in config, never from config values. Errors
after bounded retries fail the batch loudly -- no silently vectorless
rows.
- Rows without text (images, videos, failed parses) get a null vector and
are never sent to the API.
- VECTOR(dim) metadata is re-attached after the row-encoder round trip (the
encoder strips StructField metadata), so the writer detects the vector
column and the logical type lands in the committed table schema. Since the
streamer feeds only new/changed records per sync, embeddings stay current
incrementally.
Verified: transformer tests against a stub OpenAI-compatible server
(deterministic batching, null-vector rows, metadata on apply() and
transformedSchema(), 429-retry-then-success, fail-fast on client errors)
plus the streamer E2E extended to assert embedding values, upsert-refreshed
vectors, and the vector logical type in the committed schema.
* fix: use the spark adapter's row encoder for cross-version compatibility
Encoders.row(StructType) only exists on Spark 3.5+, breaking the 3.3/3.4
profile builds. SparkAdapter.getCatalystExpressionUtils().getEncoder covers
every supported Spark line.
* fix(ingest): stream unstructured rows lazily via LazyIterableIterator
Both mapPartitions sites materialized partition-scale data in Java
structures. The source built every row of a partition (inline blobs,
extracted text, chunks) into an ArrayList before returning; the embedding
transformer buffered batch.size full rows and a second same-size output
list. Both now extend hudi-common's LazyIterableIterator: the source emits
one row per file, and the transformer's only resident state is the current
embedding batch, with each buffered row released as it is emitted and the
final batch drained after input exhaustion.
Also lowers the embedding batch.size default from 1024 to 128 and documents
the two constraints it balances: per-partition transformer memory
(batch.size x average row size including inline blobs) and per-request
latency against the API timeout for single-node local endpoints.
Verified at scale: 3.68 GB / 21,790 mixed files (20k small inline, 1.5k
~1 MB inline, 150 x 8-16 MB out-of-line, binaries, PDF/DOCX) ingested via
spark-submit on Spark 4.1 in one 31.5 min sync (embedding-bound, local
Ollama), incremental sync in 25 s; row counts, blob placement, inline byte
accounting, vector dimensions and search relevance all validated exactly.
* refactor(ingest): address review on configs, parser selection and file
filtering
- document.parser enum config (TIKA default) mapping blessed names to
implementations, with CUSTOM reading parser.class as the escape hatch,
mirroring the index.type/index.class pattern
- skip columnar/data file extensions (parquet, orc, avro, hfile) by default
when no allowlist is set, so mixed directories ingest only documents
- listing.parallelism defaults to spark default parallelism instead of a
fixed 20; the config remains as an explicit cap
- new DELTA_STREAMER_TRANSFORMER config subgroup for transformer configs
- Tika 2.9.4 -> 3.3.2 (Java 11 baseline allows it)
- Lombok on ParseResult; document the on-heap inline-bytes bound
* feat(ingest): break text chunks at natural boundaries
Chunks now end at the last paragraph break, line break, sentence end or word
boundary inside the window (in that order of preference, the recursive-
splitting norm of retrieval pipelines), falling back to a hard cut for
unbroken text. A floor at half the window keeps boundary chunks from
degenerating and guarantees forward progress for any overlap setting. Chunks
remain verbatim substrings, so char_start offsets still index the original
text; unbroken-text behavior is unchanged.
* feat(ingest): pipeline embedding API requests within each partition
Keeps up to max.inflight.requests (default 2) batch requests in flight per
partition on a small daemon worker pool, hiding API latency instead of
blocking on each response; completed batches stream out row by row in input
order. Rows resident per partition are bounded by batch.size x
max.inflight.requests, documented on both configs. The HTTP client is shared
per provider instance (thread-safe, keep-alive pooled connections). Moves
EmbeddingTransformerConfig to the new transformer config subgroup and adds
direct provider unit tests: request shape, bearer auth from env, Retry-After
backoff, IOException retry, retry exhaustion, 4xx fail-fast and response
count mismatch.
* test(ingest): cover PDF, DOCX, markdown and CSV in the Tika parser tests
Adds in-line generated fixtures (minimal well-formed PDF with a text layer,
minimal OOXML DOCX zip) keeping the no-binary-files convention, with
extraction and metadata assertions. Pins commons-compress 1.27.1 in test
scope: POI needs >= 1.24 while hadoop/spark provide 1.23 on the test
classpath; Spark 4.x distributions ship a new enough version at runtime.
---
.../apache/hudi/common/config/ConfigGroups.java | 4 +
hudi-utilities/pom.xml | 24 ++
.../config/EmbeddingTransformerConfig.java | 131 ++++++++++
.../config/UnstructuredFileSourceConfig.java | 134 ++++++++++
.../sources/UnstructuredFileDFSSource.java | 184 +++++++++++++
.../helpers/unstructured/DocumentParser.java | 49 ++++
.../helpers/unstructured/DocumentParserType.java | 73 ++++++
.../sources/helpers/unstructured/ParseResult.java | 79 ++++++
.../sources/helpers/unstructured/TextChunker.java | 107 ++++++++
.../helpers/unstructured/TikaDocumentParser.java | 76 ++++++
.../UnstructuredFileRecordBuilder.java | 152 +++++++++++
.../transform/embedding/EmbeddingProvider.java | 46 ++++
.../transform/embedding/EmbeddingTransformer.java | 287 +++++++++++++++++++++
.../OpenAICompatibleEmbeddingProvider.java | 180 +++++++++++++
.../sources/TestUnstructuredFileDFSSource.java | 138 ++++++++++
.../helpers/unstructured/TestTextChunker.java | 96 +++++++
.../unstructured/TestTikaDocumentParser.java | 172 ++++++++++++
.../streamer/TestUnstructuredIngestE2E.java | 198 ++++++++++++++
.../embedding/TestEmbeddingTransformer.java | 201 +++++++++++++++
.../TestOpenAICompatibleEmbeddingProvider.java | 181 +++++++++++++
packaging/hudi-utilities-bundle/pom.xml | 1 +
packaging/hudi-utilities-slim-bundle/pom.xml | 1 +
pom.xml | 1 +
23 files changed, 2515 insertions(+)
diff --git
a/hudi-common/src/main/java/org/apache/hudi/common/config/ConfigGroups.java
b/hudi-common/src/main/java/org/apache/hudi/common/config/ConfigGroups.java
index 3ebc61a2de79..d4d2e4e679ac 100644
--- a/hudi-common/src/main/java/org/apache/hudi/common/config/ConfigGroups.java
+++ b/hudi-common/src/main/java/org/apache/hudi/common/config/ConfigGroups.java
@@ -77,6 +77,10 @@ public class ConfigGroups {
DELTA_STREAMER_SOURCE(
"Hudi Streamer Source Configs",
"Configurations controlling the behavior of reading source data."),
+ DELTA_STREAMER_TRANSFORMER(
+ "Hudi Streamer Transformer Configs",
+ "Configurations controlling the behavior of transformers applied to
source data "
+ + "before writing."),
NONE(
"None",
"No subgroup. This description should be hidden.");
diff --git a/hudi-utilities/pom.xml b/hudi-utilities/pom.xml
index 1a2a3aa877f2..5c791d8eac31 100644
--- a/hudi-utilities/pom.xml
+++ b/hudi-utilities/pom.xml
@@ -103,6 +103,30 @@
<scope>test</scope>
</dependency>
+ <!-- Apache Tika: embedded document parsing for UnstructuredFileDFSSource.
+ Only tika-core ships in the bundles; parser modules (PDF, Office,
...) are
+ supplied at runtime via spark-submit packages
(org.apache.tika:tika-parsers-standard-package) -->
+ <dependency>
+ <groupId>org.apache.tika</groupId>
+ <artifactId>tika-core</artifactId>
+ <version>${tika.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.tika</groupId>
+ <artifactId>tika-parsers-standard-package</artifactId>
+ <version>${tika.version}</version>
+ <scope>test</scope>
+ </dependency>
+ <!-- POI (Office formats via tika-parsers) needs commons-compress >= 1.24;
the
+ hadoop/spark-provided 1.23 would otherwise shadow it on the test
classpath.
+ Spark 4.x distributions already ship a new enough version at runtime.
-->
+ <dependency>
+ <groupId>org.apache.commons</groupId>
+ <artifactId>commons-compress</artifactId>
+ <version>1.27.1</version>
+ <scope>test</scope>
+ </dependency>
+
<!-- Jetty -->
<dependency>
<!-- Needs to be at the top to ensure we get the correct dependency
versions for jetty-server -->
diff --git
a/hudi-utilities/src/main/java/org/apache/hudi/utilities/config/EmbeddingTransformerConfig.java
b/hudi-utilities/src/main/java/org/apache/hudi/utilities/config/EmbeddingTransformerConfig.java
new file mode 100644
index 000000000000..6972b3917097
--- /dev/null
+++
b/hudi-utilities/src/main/java/org/apache/hudi/utilities/config/EmbeddingTransformerConfig.java
@@ -0,0 +1,131 @@
+/*
+ * 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.utilities.config;
+
+import org.apache.hudi.common.config.ConfigClassProperty;
+import org.apache.hudi.common.config.ConfigGroups;
+import org.apache.hudi.common.config.ConfigProperty;
+import org.apache.hudi.common.config.HoodieConfig;
+
+import javax.annotation.concurrent.Immutable;
+
+import static org.apache.hudi.common.util.ConfigUtils.STREAMER_CONFIG_PREFIX;
+
+/**
+ * Embedding Transformer Configs.
+ */
+@Immutable
+@ConfigClassProperty(name = "Embedding Transformer Configs",
+ groupName = ConfigGroups.Names.HUDI_STREAMER,
+ subGroupName = ConfigGroups.SubGroupNames.DELTA_STREAMER_TRANSFORMER,
+ description = "Configurations for the embedding transformer, which
populates a VECTOR "
+ + "column by calling an embedding API for each batch of ingested
records.")
+public class EmbeddingTransformerConfig extends HoodieConfig {
+
+ private static final String PREFIX = STREAMER_CONFIG_PREFIX +
"transformer.embedding.";
+
+ public static final ConfigProperty<String> PROVIDER_CLASS = ConfigProperty
+ .key(PREFIX + "provider.class")
+
.defaultValue("org.apache.hudi.utilities.transform.embedding.OpenAICompatibleEmbeddingProvider")
+ .markAdvanced()
+ .sinceVersion("1.2.0")
+ .withDocumentation("Implementation of EmbeddingProvider used to embed
record text. The "
+ + "default calls any OpenAI-compatible /v1/embeddings endpoint
(Ollama, TEI, vLLM, "
+ + "OpenAI, Voyage).");
+
+ public static final ConfigProperty<String> ENDPOINT_URL = ConfigProperty
+ .key(PREFIX + "endpoint.url")
+ .noDefaultValue()
+ .sinceVersion("1.2.0")
+ .withDocumentation("Embeddings API endpoint, e.g.
http://localhost:11434/v1/embeddings "
+ + "for a local Ollama.");
+
+ public static final ConfigProperty<String> MODEL = ConfigProperty
+ .key(PREFIX + "model")
+ .noDefaultValue()
+ .sinceVersion("1.2.0")
+ .withDocumentation("Embedding model name passed to the API, e.g.
nomic-embed-text or "
+ + "text-embedding-3-small.");
+
+ public static final ConfigProperty<String> API_KEY_ENV = ConfigProperty
+ .key(PREFIX + "api.key.env")
+ .defaultValue("")
+ .sinceVersion("1.2.0")
+ .withDocumentation("Name of the environment variable holding the API
key, sent as a "
+ + "Bearer token. Empty for unauthenticated endpoints (local
Ollama/TEI). The key "
+ + "itself is never placed in configuration.");
+
+ public static final ConfigProperty<String> DIMENSION = ConfigProperty
+ .key(PREFIX + "dimension")
+ .noDefaultValue()
+ .sinceVersion("1.2.0")
+ .withDocumentation("Dimension of the embedding vectors; declared as
VECTOR(dimension) on "
+ + "the target column and validated against API responses.");
+
+ public static final ConfigProperty<Integer> BATCH_SIZE = ConfigProperty
+ .key(PREFIX + "batch.size")
+ .defaultValue(128)
+ .markAdvanced()
+ .sinceVersion("1.2.0")
+ .withDocumentation("Number of records embedded per API request. Batching
happens at the "
+ + "record level within each Spark partition, and the batch of rows
awaiting the API "
+ + "response is the transformer's only resident state, so this also
bounds memory to "
+ + "batch.size x average row size (including inline blobs) per
partition. Raise for "
+ + "high-throughput remote APIs; lower toward 32 for single-node
local endpoints "
+ + "(Ollama, TEI) so individual requests stay well inside the request
timeout.");
+
+ public static final ConfigProperty<String> SOURCE_COLUMN = ConfigProperty
+ .key(PREFIX + "source.column")
+ .defaultValue("extracted_text")
+ .sinceVersion("1.2.0")
+ .withDocumentation("Column whose text is embedded. Rows where it is null
or empty get a "
+ + "null vector.");
+
+ public static final ConfigProperty<String> TARGET_COLUMN = ConfigProperty
+ .key(PREFIX + "target.column")
+ .defaultValue("embedding")
+ .sinceVersion("1.2.0")
+ .withDocumentation("Name of the VECTOR column appended by the
transformer.");
+
+ public static final ConfigProperty<Integer> MAX_INFLIGHT_REQUESTS =
ConfigProperty
+ .key(PREFIX + "max.inflight.requests")
+ .defaultValue(2)
+ .markAdvanced()
+ .sinceVersion("1.2.0")
+ .withDocumentation("Number of embedding API requests kept in flight per
Spark partition: "
+ + "the next batches are prefetched and sent while earlier ones
stream out, hiding API "
+ + "latency. Rows resident per partition = batch.size x this value,
so raise it only "
+ + "with the memory headroom to match.");
+
+ public static final ConfigProperty<Integer> INPUT_MAX_CHARS = ConfigProperty
+ .key(PREFIX + "input.max.chars")
+ .defaultValue(8000)
+ .markAdvanced()
+ .sinceVersion("1.2.0")
+ .withDocumentation("Input text is truncated to this many characters
before embedding, "
+ + "keeping requests inside model context limits.");
+
+ public static final ConfigProperty<Long> TIMEOUT_MS = ConfigProperty
+ .key(PREFIX + "timeout.ms")
+ .defaultValue(120_000L)
+ .markAdvanced()
+ .sinceVersion("1.2.0")
+ .withDocumentation("Per-request timeout for the embeddings API.");
+}
diff --git
a/hudi-utilities/src/main/java/org/apache/hudi/utilities/config/UnstructuredFileSourceConfig.java
b/hudi-utilities/src/main/java/org/apache/hudi/utilities/config/UnstructuredFileSourceConfig.java
new file mode 100644
index 000000000000..bd5f97ece801
--- /dev/null
+++
b/hudi-utilities/src/main/java/org/apache/hudi/utilities/config/UnstructuredFileSourceConfig.java
@@ -0,0 +1,134 @@
+/*
+ * 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.utilities.config;
+
+import org.apache.hudi.common.config.ConfigClassProperty;
+import org.apache.hudi.common.config.ConfigGroups;
+import org.apache.hudi.common.config.ConfigProperty;
+import org.apache.hudi.common.config.HoodieConfig;
+import
org.apache.hudi.utilities.sources.helpers.unstructured.DocumentParserType;
+
+import javax.annotation.concurrent.Immutable;
+
+import static org.apache.hudi.common.util.ConfigUtils.STREAMER_CONFIG_PREFIX;
+
+/**
+ * Unstructured File DFS Source Configs.
+ */
+@Immutable
+@ConfigClassProperty(name = "Unstructured File DFS Source Configs",
+ groupName = ConfigGroups.Names.HUDI_STREAMER,
+ subGroupName = ConfigGroups.SubGroupNames.DELTA_STREAMER_SOURCE,
+ description = "Configurations controlling the behavior of the unstructured
file DFS source "
+ + "in Hudi Streamer, which ingests arbitrary files (documents, images,
videos) as BLOB "
+ + "columns with extracted text, metadata and chunks.")
+public class UnstructuredFileSourceConfig extends HoodieConfig {
+
+ private static final String PREFIX = STREAMER_CONFIG_PREFIX +
"source.unstructured.";
+
+ public static final ConfigProperty<Long> BLOB_INLINE_MAX_BYTES =
ConfigProperty
+ .key(PREFIX + "blob.inline.max.bytes")
+ .defaultValue(1024L * 1024L)
+ .sinceVersion("1.2.0")
+ .withDocumentation("Files at or below this size are stored INLINE in the
blob column; "
+ + "larger files are stored OUT_OF_LINE as a reference to the
original file in place. "
+ + "This bounds per-row memory: blob bytes above the threshold never
enter Spark rows.");
+
+ public static final ConfigProperty<String> DOCUMENT_PARSER = ConfigProperty
+ .key(PREFIX + "document.parser")
+ .defaultValue("TIKA")
+ .sinceVersion("1.2.0")
+ .withDocumentation(DocumentParserType.class, "Parser used to extract
text and metadata "
+ + "from ingested files. TIKA uses Apache Tika with automatic format
detection; CUSTOM "
+ + "loads the DocumentParser implementation named by " + PREFIX +
"parser.class.");
+
+ public static final ConfigProperty<String> PARSER_CLASS = ConfigProperty
+ .key(PREFIX + "parser.class")
+ .defaultValue("")
+ .markAdvanced()
+ .sinceVersion("1.2.0")
+ .withDocumentation("Fully qualified class name of a custom
DocumentParser implementation; "
+ + "only read when " + PREFIX + "document.parser is CUSTOM.");
+
+ public static final ConfigProperty<Boolean> PARSE_ENABLED = ConfigProperty
+ .key(PREFIX + "parse.enabled")
+ .defaultValue(true)
+ .markAdvanced()
+ .sinceVersion("1.2.0")
+ .withDocumentation("When false, files are ingested as blobs only: no
text extraction, "
+ + "metadata or chunking is performed.");
+
+ public static final ConfigProperty<Long> PARSE_MAX_BYTES = ConfigProperty
+ .key(PREFIX + "parse.max.bytes")
+ .defaultValue(128L * 1024L * 1024L)
+ .markAdvanced()
+ .sinceVersion("1.2.0")
+ .withDocumentation("Files larger than this are not parsed
(parse_status=SKIPPED); they are "
+ + "still ingested as blobs. Bounds parser memory on pathological
inputs.");
+
+ public static final ConfigProperty<Integer> PARSE_MAX_TEXT_CHARS =
ConfigProperty
+ .key(PREFIX + "parse.max.text.chars")
+ .defaultValue(1_000_000)
+ .markAdvanced()
+ .sinceVersion("1.2.0")
+ .withDocumentation("Extracted text is capped at this many characters "
+ + "(parse_status=TRUNCATED when the cap is hit).");
+
+ public static final ConfigProperty<String> FILE_EXTENSIONS = ConfigProperty
+ .key(PREFIX + "file.extensions")
+ .defaultValue("")
+ .markAdvanced()
+ .sinceVersion("1.2.0")
+ .withDocumentation("Optional comma-separated allowlist of file
extensions to ingest "
+ + "(e.g. 'pdf,docx,html'). Empty ingests every file under the source
root except "
+ + "those matching " + PREFIX + "file.extensions.ignore. When set,
the allowlist "
+ + "alone decides.");
+
+ public static final ConfigProperty<String> FILE_EXTENSIONS_IGNORE =
ConfigProperty
+ .key(PREFIX + "file.extensions.ignore")
+ .defaultValue("parquet,orc,avro,hfile")
+ .markAdvanced()
+ .sinceVersion("1.2.0")
+ .withDocumentation("Comma-separated denylist of file extensions skipped
when no allowlist "
+ + "is configured. Defaults to columnar/data file formats, which
belong to structured "
+ + "sources: directories mixing data files and documents ingest only
the documents.");
+
+ public static final ConfigProperty<Integer> CHUNK_SIZE_CHARS = ConfigProperty
+ .key(PREFIX + "chunk.size.chars")
+ .defaultValue(2000)
+ .sinceVersion("1.2.0")
+ .withDocumentation("Size in characters of each text chunk emitted in the
chunks column.");
+
+ public static final ConfigProperty<Integer> CHUNK_OVERLAP_CHARS =
ConfigProperty
+ .key(PREFIX + "chunk.overlap.chars")
+ .defaultValue(200)
+ .sinceVersion("1.2.0")
+ .withDocumentation("Number of characters consecutive chunks overlap
by.");
+
+ public static final ConfigProperty<Integer> LISTING_PARALLELISM =
ConfigProperty
+ .key(PREFIX + "listing.parallelism")
+ .defaultValue(0)
+ .markAdvanced()
+ .sinceVersion("1.2.0")
+ .withDocumentation("Number of Spark partitions used to stat, fetch and
parse the files "
+ + "selected in one batch. The default 0 sizes to the cluster "
+ + "(spark default parallelism, i.e. total executor cores); set
explicitly to cap "
+ + "parse/embedding concurrency.");
+}
diff --git
a/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/UnstructuredFileDFSSource.java
b/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/UnstructuredFileDFSSource.java
new file mode 100644
index 000000000000..a6aad96fe538
--- /dev/null
+++
b/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/UnstructuredFileDFSSource.java
@@ -0,0 +1,184 @@
+/*
+ * 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.utilities.sources;
+
+import org.apache.hudi.common.config.TypedProperties;
+import org.apache.hudi.common.schema.HoodieSchema;
+import org.apache.hudi.common.schema.HoodieSchemaType;
+import org.apache.hudi.common.table.checkpoint.Checkpoint;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.common.util.collection.LazyIterableIterator;
+import org.apache.hudi.common.util.collection.Pair;
+import org.apache.hudi.hadoop.fs.HadoopFSUtils;
+import org.apache.hudi.storage.hadoop.HadoopStorageConfiguration;
+import org.apache.hudi.utilities.schema.SchemaProvider;
+import org.apache.hudi.utilities.sources.helpers.DFSPathSelector;
+import
org.apache.hudi.utilities.sources.helpers.unstructured.UnstructuredFileRecordBuilder;
+
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.spark.api.java.JavaRDD;
+import org.apache.spark.api.java.JavaSparkContext;
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Row;
+import org.apache.spark.sql.SparkSession;
+import org.apache.spark.sql.types.DataTypes;
+import org.apache.spark.sql.types.Metadata;
+import org.apache.spark.sql.types.MetadataBuilder;
+import org.apache.spark.sql.types.StructField;
+import org.apache.spark.sql.types.StructType;
+
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.util.Arrays;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Locale;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+import static org.apache.hudi.common.util.ConfigUtils.getIntWithAltKeys;
+import static org.apache.hudi.common.util.ConfigUtils.getStringWithAltKeys;
+import static
org.apache.hudi.utilities.config.UnstructuredFileSourceConfig.FILE_EXTENSIONS;
+import static
org.apache.hudi.utilities.config.UnstructuredFileSourceConfig.FILE_EXTENSIONS_IGNORE;
+import static
org.apache.hudi.utilities.config.UnstructuredFileSourceConfig.LISTING_PARALLELISM;
+
+/**
+ * DFS source that ingests unstructured files (documents, images, videos) as
rows carrying a
+ * BLOB-typed column plus extracted text, metadata and chunks.
+ *
+ * <p>File discovery and checkpointing reuse {@link DFSPathSelector}
(modification-time based,
+ * incremental). Per file, blob placement is decided by size: files at or below
+ * {@code hoodie.streamer.source.unstructured.blob.inline.max.bytes} are
stored INLINE (bytes in
+ * the table), larger files are stored OUT_OF_LINE as a reference to the
original file in place —
+ * their bytes never enter Spark rows, keeping memory and shuffle volume
bounded regardless of
+ * file sizes. Text extraction runs embedded in the executors through a
pluggable
+ * {@code DocumentParser} (Apache Tika by default); parse failures are
recorded per row and never
+ * fail the ingestion job.
+ *
+ * <p>Keying the table on {@code path} with ordering on {@code
modification_time} makes
+ * re-ingested files upsert in place, so downstream text (and any embedding
columns added by
+ * transformers) stay current with the source directory.
+ */
+public class UnstructuredFileDFSSource extends RowSource {
+
+ private static final Metadata BLOB_METADATA = new MetadataBuilder()
+ .putString(HoodieSchema.TYPE_METADATA_FIELD,
HoodieSchemaType.BLOB.name())
+ .build();
+
+ private static final StructType BLOB_REFERENCE_TYPE =
DataTypes.createStructType(new StructField[] {
+ DataTypes.createStructField(HoodieSchema.Blob.EXTERNAL_REFERENCE_PATH,
DataTypes.StringType, true),
+ DataTypes.createStructField(HoodieSchema.Blob.EXTERNAL_REFERENCE_OFFSET,
DataTypes.LongType, true),
+ DataTypes.createStructField(HoodieSchema.Blob.EXTERNAL_REFERENCE_LENGTH,
DataTypes.LongType, true),
+
DataTypes.createStructField(HoodieSchema.Blob.EXTERNAL_REFERENCE_IS_MANAGED,
DataTypes.BooleanType, true)});
+
+ private static final StructType BLOB_TYPE = DataTypes.createStructType(new
StructField[] {
+ DataTypes.createStructField(HoodieSchema.Blob.TYPE,
DataTypes.StringType, false),
+ DataTypes.createStructField(HoodieSchema.Blob.INLINE_DATA_FIELD,
DataTypes.BinaryType, true),
+ DataTypes.createStructField(HoodieSchema.Blob.EXTERNAL_REFERENCE,
BLOB_REFERENCE_TYPE, true)});
+
+ private static final StructType CHUNK_TYPE = DataTypes.createStructType(new
StructField[] {
+ DataTypes.createStructField("chunk_id", DataTypes.IntegerType, false),
+ DataTypes.createStructField("text", DataTypes.StringType, false),
+ DataTypes.createStructField("char_start", DataTypes.IntegerType,
false)});
+
+ public static final StructType SOURCE_SCHEMA = new StructType(new
StructField[] {
+ DataTypes.createStructField("path", DataTypes.StringType, false),
+ DataTypes.createStructField("file_name", DataTypes.StringType, false),
+ DataTypes.createStructField("extension", DataTypes.StringType, false),
+ DataTypes.createStructField("size", DataTypes.LongType, false),
+ DataTypes.createStructField("modification_time", DataTypes.LongType,
false),
+ new StructField("content", BLOB_TYPE, false, BLOB_METADATA),
+ DataTypes.createStructField("extracted_text", DataTypes.StringType,
true),
+ DataTypes.createStructField("doc_metadata",
+ DataTypes.createMapType(DataTypes.StringType, DataTypes.StringType,
true), true),
+ DataTypes.createStructField("chunks",
DataTypes.createArrayType(CHUNK_TYPE, false), true),
+ DataTypes.createStructField("parse_status", DataTypes.StringType, false),
+ DataTypes.createStructField("parse_error", DataTypes.StringType, true)});
+
+ private final DFSPathSelector pathSelector;
+ private final UnstructuredFileRecordBuilder recordBuilder;
+ private final Set<String> allowedExtensions;
+ private final Set<String> ignoredExtensions;
+ private final int listingParallelism;
+
+ public UnstructuredFileDFSSource(TypedProperties props, JavaSparkContext
sparkContext, SparkSession sparkSession,
+ SchemaProvider schemaProvider) {
+ super(props, sparkContext, sparkSession, schemaProvider);
+ this.pathSelector = DFSPathSelector.createSourceSelector(props,
sparkContext.hadoopConfiguration());
+ this.recordBuilder = new UnstructuredFileRecordBuilder(props);
+ this.allowedExtensions = parseExtensions(getStringWithAltKeys(props,
FILE_EXTENSIONS, true));
+ this.ignoredExtensions = parseExtensions(getStringWithAltKeys(props,
FILE_EXTENSIONS_IGNORE, true));
+ int configuredParallelism = getIntWithAltKeys(props, LISTING_PARALLELISM);
+ this.listingParallelism = configuredParallelism > 0
+ ? configuredParallelism : sparkContext.defaultParallelism();
+ }
+
+ private static Set<String> parseExtensions(String csv) {
+ return csv == null || csv.trim().isEmpty()
+ ? new HashSet<>()
+ : Arrays.stream(csv.toLowerCase(Locale.ROOT).split(","))
+ .map(String::trim).filter(s ->
!s.isEmpty()).collect(Collectors.toSet());
+ }
+
+ private boolean isEligible(String fileName) {
+ String extension = UnstructuredFileRecordBuilder.extensionOf(fileName);
+ // an explicit allowlist decides alone; otherwise everything except the
denylist
+ return allowedExtensions.isEmpty()
+ ? !ignoredExtensions.contains(extension) :
allowedExtensions.contains(extension);
+ }
+
+ @Override
+ public Pair<Option<Dataset<Row>>, Checkpoint>
fetchNextBatch(Option<Checkpoint> lastCheckpoint, long sourceLimit) {
+ Pair<Option<String>, Checkpoint> selected =
+ pathSelector.getNextFilePathsAndMaxModificationTime(sparkContext,
lastCheckpoint, sourceLimit);
+ return selected.getLeft()
+ .map(pathStr -> Pair.of(Option.of(fromFiles(pathStr)),
selected.getRight()))
+ .orElseGet(() -> Pair.of(Option.empty(), selected.getRight()));
+ }
+
+ private Dataset<Row> fromFiles(String pathStr) {
+ List<String> paths = Arrays.stream(pathStr.split(","))
+ .filter(p -> isEligible(new Path(p).getName()))
+ .collect(Collectors.toList());
+ int parallelism = Math.max(1, Math.min(paths.size(), listingParallelism));
+ HadoopStorageConfiguration storageConf = new
HadoopStorageConfiguration(sparkContext.hadoopConfiguration());
+ UnstructuredFileRecordBuilder builder = this.recordBuilder;
+ // one file -> one row, lazily: inline bytes and extracted text of a
partition must
+ // never be resident all at once, or partition memory scales with total
corpus size
+ JavaRDD<Row> rows = sparkContext.parallelize(paths,
parallelism).mapPartitions(pathIterator ->
+ new LazyIterableIterator<String, Row>(pathIterator) {
+ private FileSystem fs;
+
+ @Override
+ protected Row computeNext() {
+ String path = inputItr.next();
+ try {
+ if (fs == null) {
+ fs = HadoopFSUtils.getFs(new Path(path), storageConf);
+ }
+ return builder.buildRow(fs, path);
+ } catch (IOException e) {
+ throw new UncheckedIOException("Failed to build record for " +
path, e);
+ }
+ }
+ });
+ return sparkSession.createDataFrame(rows, SOURCE_SCHEMA);
+ }
+}
diff --git
a/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/helpers/unstructured/DocumentParser.java
b/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/helpers/unstructured/DocumentParser.java
new file mode 100644
index 000000000000..a71cd96361b6
--- /dev/null
+++
b/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/helpers/unstructured/DocumentParser.java
@@ -0,0 +1,49 @@
+/*
+ * 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.utilities.sources.helpers.unstructured;
+
+import org.apache.hudi.common.config.TypedProperties;
+
+import java.io.InputStream;
+import java.io.Serializable;
+
+/**
+ * Extracts text and metadata from a single file's content stream.
Implementations run
+ * inside Spark executors, must be embedded JVM libraries (no external service
calls),
+ * and must NEVER throw from {@link #parse}: any failure is reported via
+ * {@link ParseResult#failed}.
+ */
+public interface DocumentParser extends Serializable {
+
+ /**
+ * Called once per executor instance before the first {@link #parse}.
+ */
+ default void init(TypedProperties props) {
+ }
+
+ /**
+ * Parses one file.
+ *
+ * @param in content stream; the caller owns closing it
+ * @param fileName name of the file (used for format detection hints)
+ * @param maxTextChars cap on extracted text length; hitting it yields a
TRUNCATED result
+ */
+ ParseResult parse(InputStream in, String fileName, int maxTextChars);
+}
diff --git
a/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/helpers/unstructured/DocumentParserType.java
b/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/helpers/unstructured/DocumentParserType.java
new file mode 100644
index 000000000000..5279b99228a5
--- /dev/null
+++
b/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/helpers/unstructured/DocumentParserType.java
@@ -0,0 +1,73 @@
+/*
+ * 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.utilities.sources.helpers.unstructured;
+
+import org.apache.hudi.common.config.EnumDescription;
+import org.apache.hudi.common.config.EnumFieldDescription;
+import org.apache.hudi.common.config.TypedProperties;
+import org.apache.hudi.exception.HoodieException;
+
+import java.util.Locale;
+
+import static org.apache.hudi.common.util.ConfigUtils.getStringWithAltKeys;
+import static
org.apache.hudi.utilities.config.UnstructuredFileSourceConfig.DOCUMENT_PARSER;
+import static
org.apache.hudi.utilities.config.UnstructuredFileSourceConfig.PARSER_CLASS;
+
+/**
+ * Blessed {@link DocumentParser} implementations selectable by name, with
CUSTOM as the
+ * escape hatch for user-supplied classes (mirrors the index.type /
index.class pattern).
+ */
+@EnumDescription("Parser used to extract text and metadata from ingested
unstructured files.")
+public enum DocumentParserType {
+
+ @EnumFieldDescription("Apache Tika AutoDetectParser: in-process extraction
for 1000+ formats.")
+
TIKA("org.apache.hudi.utilities.sources.helpers.unstructured.TikaDocumentParser"),
+
+ @EnumFieldDescription("Load the DocumentParser implementation named by "
+ + "hoodie.streamer.source.unstructured.parser.class.")
+ CUSTOM(null);
+
+ private final String parserClassName;
+
+ DocumentParserType(String parserClassName) {
+ this.parserClassName = parserClassName;
+ }
+
+ /**
+ * Resolves the configured parser type to the class name to instantiate.
+ */
+ public static String resolveParserClass(TypedProperties props) {
+ String type = getStringWithAltKeys(props, DOCUMENT_PARSER, true);
+ DocumentParserType parserType;
+ try {
+ parserType =
DocumentParserType.valueOf(type.trim().toUpperCase(Locale.ROOT));
+ } catch (IllegalArgumentException e) {
+ throw new HoodieException("Unknown " + DOCUMENT_PARSER.key() + " value:
" + type);
+ }
+ if (parserType != CUSTOM) {
+ return parserType.parserClassName;
+ }
+ String customClass = getStringWithAltKeys(props, PARSER_CLASS, true);
+ if (customClass == null || customClass.trim().isEmpty()) {
+ throw new HoodieException(DOCUMENT_PARSER.key() + "=CUSTOM requires " +
PARSER_CLASS.key());
+ }
+ return customClass.trim();
+ }
+}
diff --git
a/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/helpers/unstructured/ParseResult.java
b/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/helpers/unstructured/ParseResult.java
new file mode 100644
index 000000000000..1f91b3ab2a3a
--- /dev/null
+++
b/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/helpers/unstructured/ParseResult.java
@@ -0,0 +1,79 @@
+/*
+ * 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.utilities.sources.helpers.unstructured;
+
+import lombok.Getter;
+
+import java.io.Serializable;
+import java.util.Collections;
+import java.util.Map;
+
+/**
+ * Outcome of parsing one file with a {@link DocumentParser}.
+ */
+@Getter
+public class ParseResult implements Serializable {
+
+ private static final long serialVersionUID = 1L;
+
+ /**
+ * Row-level parse outcome. Parsing never fails the ingestion job; failures
are
+ * recorded on the row so consumers can filter or reprocess.
+ */
+ public enum ParseStatus {
+ /** Text extracted in full. */
+ SUCCESS,
+ /** Text extracted but capped at the configured character limit. */
+ TRUNCATED,
+ /** Parsed without error but no text content (e.g. images, videos, unknown
formats). */
+ EMPTY,
+ /** File exceeded the parse size cap or its extension/format is not
parseable; blob still ingested. */
+ SKIPPED,
+ /** Parser threw; error message recorded, blob still ingested. */
+ FAILED;
+ }
+
+ private final ParseStatus status;
+ private final String text;
+ private final Map<String, String> metadata;
+ private final String error;
+
+ private ParseResult(ParseStatus status, String text, Map<String, String>
metadata, String error) {
+ this.status = status;
+ this.text = text;
+ this.metadata = metadata == null ? Collections.emptyMap() : metadata;
+ this.error = error;
+ }
+
+ public static ParseResult success(String text, Map<String, String> metadata,
boolean truncated) {
+ if (text == null || text.trim().isEmpty()) {
+ return new ParseResult(ParseStatus.EMPTY, "", metadata, null);
+ }
+ return new ParseResult(truncated ? ParseStatus.TRUNCATED :
ParseStatus.SUCCESS, text, metadata, null);
+ }
+
+ public static ParseResult skipped(String reason) {
+ return new ParseResult(ParseStatus.SKIPPED, "", Collections.emptyMap(),
reason);
+ }
+
+ public static ParseResult failed(String error) {
+ return new ParseResult(ParseStatus.FAILED, "", Collections.emptyMap(),
error);
+ }
+}
diff --git
a/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/helpers/unstructured/TextChunker.java
b/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/helpers/unstructured/TextChunker.java
new file mode 100644
index 000000000000..2b672f896ace
--- /dev/null
+++
b/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/helpers/unstructured/TextChunker.java
@@ -0,0 +1,107 @@
+/*
+ * 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.utilities.sources.helpers.unstructured;
+
+import org.apache.hudi.common.util.ValidationUtils;
+
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
+/**
+ * Splits extracted text into overlapping chunks for retrieval-oriented
consumers,
+ * breaking at natural boundaries where possible: each chunk is at most
+ * {@code chunk.size.chars} long and ends at the last paragraph break, line
break,
+ * sentence end or word boundary inside its window (in that order of
preference,
+ * the recursive-splitting norm of retrieval pipelines), falling back to a
hard cut
+ * for unbroken text. Chunks are verbatim substrings so {@code char_start}
offsets
+ * always index into the original text.
+ */
+public class TextChunker implements Serializable {
+
+ private static final long serialVersionUID = 1L;
+
+ // in order of preference; a break lands just after the matched separator
+ private static final String[] BOUNDARIES = {"\n\n", "\n", ". ", "! ", "? ",
" "};
+
+ private final int chunkSizeChars;
+ private final int overlapChars;
+
+ public TextChunker(int chunkSizeChars, int overlapChars) {
+ ValidationUtils.checkArgument(chunkSizeChars > 0, "chunk size must be
positive");
+ ValidationUtils.checkArgument(overlapChars >= 0 && overlapChars <
chunkSizeChars,
+ "chunk overlap must be non-negative and smaller than the chunk size");
+ this.chunkSizeChars = chunkSizeChars;
+ this.overlapChars = overlapChars;
+ }
+
+ /**
+ * One chunk of text and its position within the source document.
+ */
+ public static class Chunk implements Serializable {
+ private static final long serialVersionUID = 1L;
+
+ public final int chunkId;
+ public final String text;
+ public final int charStart;
+
+ public Chunk(int chunkId, String text, int charStart) {
+ this.chunkId = chunkId;
+ this.text = text;
+ this.charStart = charStart;
+ }
+ }
+
+ public List<Chunk> chunk(String text) {
+ if (text == null || text.isEmpty()) {
+ return Collections.emptyList();
+ }
+ List<Chunk> chunks = new ArrayList<>();
+ int start = 0;
+ for (int id = 0; start < text.length(); id++) {
+ int windowEnd = Math.min(start + chunkSizeChars, text.length());
+ int end = windowEnd == text.length() ? windowEnd : findBreak(text,
start, windowEnd);
+ chunks.add(new Chunk(id, text.substring(start, end), start));
+ if (end == text.length()) {
+ break;
+ }
+ start = end - overlapChars;
+ }
+ return chunks;
+ }
+
+ /**
+ * Best break position in {@code (minBreak, windowEnd]}, preferring the
strongest
+ * boundary. The floor guarantees forward progress after overlap is
subtracted and
+ * keeps boundary chunks from degenerating below half the window.
+ */
+ private int findBreak(String text, int start, int windowEnd) {
+ int minBreak = start + Math.max(overlapChars + 1, chunkSizeChars / 2);
+ for (String boundary : BOUNDARIES) {
+ int idx = text.lastIndexOf(boundary, windowEnd - boundary.length());
+ int breakEnd = idx + boundary.length();
+ if (idx >= 0 && breakEnd > minBreak && breakEnd <= windowEnd) {
+ return breakEnd;
+ }
+ }
+ return windowEnd;
+ }
+}
diff --git
a/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/helpers/unstructured/TikaDocumentParser.java
b/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/helpers/unstructured/TikaDocumentParser.java
new file mode 100644
index 000000000000..e44f7a7e99e5
--- /dev/null
+++
b/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/helpers/unstructured/TikaDocumentParser.java
@@ -0,0 +1,76 @@
+/*
+ * 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.utilities.sources.helpers.unstructured;
+
+import org.apache.tika.exception.WriteLimitReachedException;
+import org.apache.tika.metadata.Metadata;
+import org.apache.tika.metadata.TikaCoreProperties;
+import org.apache.tika.parser.AutoDetectParser;
+import org.apache.tika.parser.ParseContext;
+import org.apache.tika.sax.BodyContentHandler;
+
+import java.io.InputStream;
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * Default {@link DocumentParser} backed by Apache Tika's {@link
AutoDetectParser}:
+ * in-process text and metadata extraction for 1000+ formats (PDF via PDFBox,
Office
+ * via POI, HTML, plain text, image/video container metadata). Which formats
parse
+ * depends on the Tika parser modules present on the classpath; with only
tika-core
+ * available, non-plain-text formats yield EMPTY results rather than errors.
+ */
+public class TikaDocumentParser implements DocumentParser {
+
+ private static final long serialVersionUID = 1L;
+
+ private transient AutoDetectParser parser;
+
+ @Override
+ public ParseResult parse(InputStream in, String fileName, int maxTextChars) {
+ try {
+ if (parser == null) {
+ parser = new AutoDetectParser();
+ }
+ BodyContentHandler handler = new BodyContentHandler(maxTextChars);
+ Metadata tikaMetadata = new Metadata();
+ tikaMetadata.set(TikaCoreProperties.RESOURCE_NAME_KEY, fileName);
+ boolean truncated = false;
+ try {
+ parser.parse(in, handler, tikaMetadata, new ParseContext());
+ } catch (WriteLimitReachedException e) {
+ // Hitting the char cap is a successful-but-truncated parse, not a
failure.
+ truncated = true;
+ }
+ return ParseResult.success(handler.toString().trim(),
toMap(tikaMetadata), truncated);
+ } catch (Throwable t) {
+ // Never propagate: a corrupt file must not fail the ingestion job.
+ return ParseResult.failed(t.getClass().getSimpleName() + ": " +
String.valueOf(t.getMessage()));
+ }
+ }
+
+ private static Map<String, String> toMap(Metadata metadata) {
+ Map<String, String> map = new HashMap<>();
+ for (String name : metadata.names()) {
+ map.put(name, metadata.get(name));
+ }
+ return map;
+ }
+}
diff --git
a/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/helpers/unstructured/UnstructuredFileRecordBuilder.java
b/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/helpers/unstructured/UnstructuredFileRecordBuilder.java
new file mode 100644
index 000000000000..0ac00ea4f316
--- /dev/null
+++
b/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/helpers/unstructured/UnstructuredFileRecordBuilder.java
@@ -0,0 +1,152 @@
+/*
+ * 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.utilities.sources.helpers.unstructured;
+
+import org.apache.hudi.common.config.TypedProperties;
+import org.apache.hudi.common.schema.HoodieSchema;
+import org.apache.hudi.common.util.ReflectionUtils;
+
+import org.apache.hadoop.fs.FSDataInputStream;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.spark.sql.Row;
+import org.apache.spark.sql.RowFactory;
+
+import java.io.ByteArrayInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.Serializable;
+import java.util.List;
+import java.util.stream.Collectors;
+
+import static org.apache.hudi.common.util.ConfigUtils.getBooleanWithAltKeys;
+import static org.apache.hudi.common.util.ConfigUtils.getIntWithAltKeys;
+import static org.apache.hudi.common.util.ConfigUtils.getLongWithAltKeys;
+import static
org.apache.hudi.utilities.config.UnstructuredFileSourceConfig.BLOB_INLINE_MAX_BYTES;
+import static
org.apache.hudi.utilities.config.UnstructuredFileSourceConfig.CHUNK_OVERLAP_CHARS;
+import static
org.apache.hudi.utilities.config.UnstructuredFileSourceConfig.CHUNK_SIZE_CHARS;
+import static
org.apache.hudi.utilities.config.UnstructuredFileSourceConfig.PARSE_ENABLED;
+import static
org.apache.hudi.utilities.config.UnstructuredFileSourceConfig.PARSE_MAX_BYTES;
+import static
org.apache.hudi.utilities.config.UnstructuredFileSourceConfig.PARSE_MAX_TEXT_CHARS;
+
+/**
+ * Executor-side logic turning one file into one source {@link Row}: decides
inline vs
+ * out-of-line blob placement by size, fetches bytes only for inline files,
parses and
+ * chunks content. Blob bytes above the inline threshold never enter Spark
rows — the
+ * blob column carries a reference to the original file in place.
+ */
+public class UnstructuredFileRecordBuilder implements Serializable {
+
+ private static final long serialVersionUID = 1L;
+
+ private final long inlineMaxBytes;
+ private final boolean parseEnabled;
+ private final long parseMaxBytes;
+ private final int parseMaxTextChars;
+ private final String parserClass;
+ private final TextChunker chunker;
+ private final TypedProperties props;
+
+ private transient DocumentParser parser;
+
+ public UnstructuredFileRecordBuilder(TypedProperties props) {
+ this.props = props;
+ this.inlineMaxBytes = getLongWithAltKeys(props, BLOB_INLINE_MAX_BYTES);
+ this.parseEnabled = getBooleanWithAltKeys(props, PARSE_ENABLED);
+ this.parseMaxBytes = getLongWithAltKeys(props, PARSE_MAX_BYTES);
+ this.parseMaxTextChars = getIntWithAltKeys(props, PARSE_MAX_TEXT_CHARS);
+ this.parserClass = DocumentParserType.resolveParserClass(props);
+ this.chunker = new TextChunker(getIntWithAltKeys(props, CHUNK_SIZE_CHARS),
+ getIntWithAltKeys(props, CHUNK_OVERLAP_CHARS));
+ }
+
+ public Row buildRow(FileSystem fs, String pathStr) throws IOException {
+ Path path = new Path(pathStr);
+ FileStatus status = fs.getFileStatus(path);
+ long size = status.getLen();
+ String fileName = path.getName();
+
+ byte[] inlineBytes = null;
+ Row blob;
+ if (size <= inlineMaxBytes) {
+ inlineBytes = readFully(fs, path, (int) size);
+ blob = RowFactory.create(HoodieSchema.Blob.INLINE, inlineBytes, null);
+ } else {
+ Row reference = RowFactory.create(pathStr, null, null, false);
+ blob = RowFactory.create(HoodieSchema.Blob.OUT_OF_LINE, null, reference);
+ }
+
+ ParseResult parseResult = parse(fs, path, fileName, size, inlineBytes);
+ List<Row> chunks = chunker.chunk(parseResult.getText()).stream()
+ .map(c -> RowFactory.create(c.chunkId, c.text, c.charStart))
+ .collect(Collectors.toList());
+
+ return RowFactory.create(
+ pathStr,
+ fileName,
+ extensionOf(fileName),
+ size,
+ status.getModificationTime(),
+ blob,
+ parseResult.getText(),
+ // Spark's Row encoder requires a scala Map as the external type for
MapType
+
scala.collection.JavaConverters.mapAsScalaMapConverter(parseResult.getMetadata()).asScala(),
+ chunks.toArray(new Row[0]),
+ parseResult.getStatus().name(),
+ parseResult.getError());
+ }
+
+ private ParseResult parse(FileSystem fs, Path path, String fileName, long
size, byte[] inlineBytes) {
+ if (!parseEnabled) {
+ return ParseResult.skipped("parsing disabled");
+ }
+ if (size > parseMaxBytes) {
+ return ParseResult.skipped("file size " + size + " exceeds " +
PARSE_MAX_BYTES.key());
+ }
+ if (parser == null) {
+ parser = (DocumentParser) ReflectionUtils.loadClass(parserClass);
+ parser.init(props);
+ }
+ try (InputStream in = inlineBytes != null
+ ? new ByteArrayInputStream(inlineBytes) : fs.open(path)) {
+ return parser.parse(in, fileName, parseMaxTextChars);
+ } catch (Exception e) {
+ return ParseResult.failed(e.getClass().getSimpleName() + ": " +
e.getMessage());
+ }
+ }
+
+ private static byte[] readFully(FileSystem fs, Path path, int size) throws
IOException {
+ // On-heap by design: Spark's BinaryType external type is byte[], so an
off-heap read
+ // would still copy onto the heap to build the row. Peak per-task
allocation is bounded
+ // by BLOB_INLINE_MAX_BYTES (rows stream one at a time); raising that
threshold raises
+ // task memory accordingly.
+ byte[] bytes = new byte[size];
+ try (FSDataInputStream in = fs.open(path)) {
+ in.readFully(0, bytes);
+ }
+ return bytes;
+ }
+
+ public static String extensionOf(String fileName) {
+ int dot = fileName.lastIndexOf('.');
+ return dot > 0 && dot < fileName.length() - 1 ? fileName.substring(dot +
1).toLowerCase() : "";
+ }
+}
diff --git
a/hudi-utilities/src/main/java/org/apache/hudi/utilities/transform/embedding/EmbeddingProvider.java
b/hudi-utilities/src/main/java/org/apache/hudi/utilities/transform/embedding/EmbeddingProvider.java
new file mode 100644
index 000000000000..22f625821383
--- /dev/null
+++
b/hudi-utilities/src/main/java/org/apache/hudi/utilities/transform/embedding/EmbeddingProvider.java
@@ -0,0 +1,46 @@
+/*
+ * 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.utilities.transform.embedding;
+
+import org.apache.hudi.common.config.TypedProperties;
+
+import java.io.Serializable;
+import java.util.List;
+
+/**
+ * Produces embedding vectors for batches of texts. Implementations run inside
Spark
+ * executors and are called with record-level batches buffered within a
partition; an
+ * in-JVM (e.g. ONNX) implementation can slot in beside API-backed ones.
+ */
+public interface EmbeddingProvider extends Serializable {
+
+ /**
+ * Called once per executor instance before the first {@link #embed}.
+ */
+ default void init(TypedProperties props) {
+ }
+
+ /**
+ * Embeds a batch of texts, returning one vector per input, in order.
+ * Errors should be retried internally where transient; a thrown exception
+ * fails the batch (and the sync) -- the caller never silently drops vectors.
+ */
+ List<float[]> embed(List<String> texts);
+}
diff --git
a/hudi-utilities/src/main/java/org/apache/hudi/utilities/transform/embedding/EmbeddingTransformer.java
b/hudi-utilities/src/main/java/org/apache/hudi/utilities/transform/embedding/EmbeddingTransformer.java
new file mode 100644
index 000000000000..1fff79426d23
--- /dev/null
+++
b/hudi-utilities/src/main/java/org/apache/hudi/utilities/transform/embedding/EmbeddingTransformer.java
@@ -0,0 +1,287 @@
+/*
+ * 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.utilities.transform.embedding;
+
+import org.apache.hudi.SparkAdapterSupport$;
+import org.apache.hudi.common.config.TypedProperties;
+import org.apache.hudi.common.schema.HoodieSchema;
+import org.apache.hudi.common.util.CustomizedThreadFactory;
+import org.apache.hudi.common.util.ReflectionUtils;
+import org.apache.hudi.common.util.collection.LazyIterableIterator;
+import org.apache.hudi.exception.HoodieException;
+import org.apache.hudi.utilities.transform.Transformer;
+
+import org.apache.spark.api.java.JavaSparkContext;
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Row;
+import org.apache.spark.sql.RowFactory;
+import org.apache.spark.sql.SparkSession;
+import org.apache.spark.sql.types.ArrayType;
+import org.apache.spark.sql.types.DataTypes;
+import org.apache.spark.sql.types.Metadata;
+import org.apache.spark.sql.types.MetadataBuilder;
+import org.apache.spark.sql.types.StructField;
+import org.apache.spark.sql.types.StructType;
+
+import java.util.ArrayDeque;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+
+import static org.apache.hudi.common.util.ConfigUtils.getIntWithAltKeys;
+import static org.apache.hudi.common.util.ConfigUtils.getStringWithAltKeys;
+import static
org.apache.hudi.utilities.config.EmbeddingTransformerConfig.BATCH_SIZE;
+import static
org.apache.hudi.utilities.config.EmbeddingTransformerConfig.DIMENSION;
+import static
org.apache.hudi.utilities.config.EmbeddingTransformerConfig.INPUT_MAX_CHARS;
+import static
org.apache.hudi.utilities.config.EmbeddingTransformerConfig.MAX_INFLIGHT_REQUESTS;
+import static
org.apache.hudi.utilities.config.EmbeddingTransformerConfig.PROVIDER_CLASS;
+import static
org.apache.hudi.utilities.config.EmbeddingTransformerConfig.SOURCE_COLUMN;
+import static
org.apache.hudi.utilities.config.EmbeddingTransformerConfig.TARGET_COLUMN;
+
+/**
+ * Appends a VECTOR(dimension) embedding column by calling an embedding API
for the text
+ * in {@code source.column}. Batching happens at the record level within each
partition:
+ * up to {@code batch.size} records' texts go into one API request, and the
batch is
+ * streamed back out row by row before the next one is pulled, so {@code
batch.size}
+ * bounds the rows resident per partition. Retry and backoff (in the provider)
are the
+ * only flow control. Rows with no text (e.g. images, videos, failed parses)
receive a
+ * null vector and are never sent to the API.
+ *
+ * <p>Because the streamer only feeds new or changed records through the
transformer chain
+ * each sync, embeddings stay current with the ingested data at no extra cost.
+ */
+public class EmbeddingTransformer implements Transformer {
+
+ @Override
+ public Dataset<Row> apply(JavaSparkContext jsc, SparkSession sparkSession,
Dataset<Row> rowDataset,
+ TypedProperties properties) {
+ String sourceColumn = getStringWithAltKeys(properties, SOURCE_COLUMN,
true);
+ String targetColumn = getStringWithAltKeys(properties, TARGET_COLUMN,
true);
+ int dimension = getIntWithAltKeys(properties, DIMENSION);
+ int batchSize = getIntWithAltKeys(properties, BATCH_SIZE);
+ int inputMaxChars = getIntWithAltKeys(properties, INPUT_MAX_CHARS);
+ int maxInflight = getIntWithAltKeys(properties, MAX_INFLIGHT_REQUESTS);
+ String providerClass = getStringWithAltKeys(properties, PROVIDER_CLASS,
true);
+
+ StructType inputSchema = rowDataset.schema();
+ int sourceIndex = inputSchema.fieldIndex(sourceColumn);
+ StructType outputSchema = withVectorColumn(inputSchema, targetColumn,
dimension);
+
+ // Encoders.row(schema) only exists on Spark 3.5+; the adapter covers
3.3/3.4/4.x too
+ Dataset<Row> withVectors = rowDataset.mapPartitions(
+ (org.apache.spark.api.java.function.MapPartitionsFunction<Row, Row>)
partition ->
+ new EmbeddingIterator(partition, providerClass, properties,
sourceIndex,
+ dimension, batchSize, inputMaxChars, maxInflight),
+
SparkAdapterSupport$.MODULE$.sparkAdapter().getCatalystExpressionUtils().getEncoder(outputSchema));
+ // the row encoder drops StructField metadata; re-attach VECTOR(dim) so the
+ // writer detects the column (and StreamSync deduces the right target
schema)
+ Metadata vectorMetadata = outputSchema.apply(targetColumn).metadata();
+ return withVectors.withColumn(targetColumn,
+ withVectors.col(targetColumn).as(targetColumn, vectorMetadata));
+ }
+
+ static StructType withVectorColumn(StructType schema, String targetColumn,
int dimension) {
+ Metadata vectorMetadata = new MetadataBuilder()
+ .putString(HoodieSchema.TYPE_METADATA_FIELD, "VECTOR(" + dimension +
")")
+ .build();
+ ArrayType vectorType = DataTypes.createArrayType(DataTypes.FloatType,
false);
+ return schema.add(new StructField(targetColumn, vectorType, true,
vectorMetadata));
+ }
+
+ /**
+ * Pulls up to {@code batch.size} input rows per batch, keeps up to
+ * {@code max.inflight.requests} batches' API calls in flight on a small
worker pool,
+ * and streams each completed batch out row by row (releasing every buffered
row as it
+ * is emitted) in input order. Rows resident per partition are bounded by
+ * batch.size x max.inflight.requests.
+ */
+ private static class EmbeddingIterator extends LazyIterableIterator<Row,
Row> {
+
+ private final String providerClass;
+ private final TypedProperties props;
+ private final int sourceIndex;
+ private final int dimension;
+ private final int batchSize;
+ private final int inputMaxChars;
+ private final int maxInflight;
+
+ private EmbeddingProvider provider;
+ private ExecutorService executor;
+ private final ArrayDeque<PendingBatch> inflight = new ArrayDeque<>();
+ private Row[] batch;
+ private List<Float>[] batchVectors;
+ private int batchCount;
+ private int emitIndex;
+
+ EmbeddingIterator(Iterator<Row> input, String providerClass,
TypedProperties props,
+ int sourceIndex, int dimension, int batchSize, int inputMaxChars, int
maxInflight) {
+ super(input);
+ this.providerClass = providerClass;
+ this.props = props;
+ this.sourceIndex = sourceIndex;
+ this.dimension = dimension;
+ this.batchSize = batchSize;
+ this.inputMaxChars = inputMaxChars;
+ this.maxInflight = maxInflight;
+ }
+
+ /**
+ * One batch of buffered rows whose embedding request is submitted but not
yet drained.
+ */
+ private static class PendingBatch {
+ final Row[] rows;
+ final int count;
+ final List<Integer> textRowIndexes;
+ final Future<List<float[]>> vectors;
+
+ PendingBatch(Row[] rows, int count, List<Integer> textRowIndexes,
Future<List<float[]>> vectors) {
+ this.rows = rows;
+ this.count = count;
+ this.textRowIndexes = textRowIndexes;
+ this.vectors = vectors;
+ }
+ }
+
+ @Override
+ public boolean hasNext() {
+ // drain buffered and in-flight batches before consulting the input; the
+ // short-circuit keeps super.hasNext() (and its end() hook) from firing
early
+ return emitIndex < batchCount || !inflight.isEmpty() || super.hasNext();
+ }
+
+ @Override
+ protected Row computeNext() {
+ if (emitIndex >= batchCount) {
+ promoteNextBatch();
+ }
+ Row row = batch[emitIndex];
+ List<Float> vector = batchVectors[emitIndex];
+ batch[emitIndex] = null;
+ batchVectors[emitIndex] = null;
+ emitIndex++;
+
+ Object[] values = new Object[row.length() + 1];
+ for (int f = 0; f < row.length(); f++) {
+ values[f] = row.get(f);
+ }
+ // the Row encoder expects a scala Seq as the external type for array
columns
+ values[row.length()] = vector == null
+ ? null : scala.collection.JavaConverters.asScalaBuffer(vector);
+ return RowFactory.create(values);
+ }
+
+ @Override
+ protected void end() {
+ if (executor != null) {
+ executor.shutdownNow();
+ }
+ }
+
+ /**
+ * Tops the in-flight window up, then blocks on the oldest batch's
response and makes
+ * it the draining batch. Batch order equals input order.
+ */
+ @SuppressWarnings("unchecked")
+ private void promoteNextBatch() {
+ submitUpToWindow();
+ PendingBatch pending = inflight.poll();
+ submitUpToWindow();
+
+ List<float[]> vectors;
+ try {
+ vectors = pending.vectors.get();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new HoodieException("Interrupted waiting for embeddings
response", e);
+ } catch (ExecutionException e) {
+ throw e.getCause() instanceof HoodieException
+ ? (HoodieException) e.getCause()
+ : new HoodieException("Embedding request failed", e.getCause());
+ }
+
+ batch = pending.rows;
+ batchCount = pending.count;
+ batchVectors = new List[batchCount];
+ emitIndex = 0;
+ for (int i = 0; i < vectors.size(); i++) {
+ float[] vector = vectors.get(i);
+ if (vector.length != dimension) {
+ throw new HoodieException("Embeddings API returned dimension " +
vector.length
+ + " but " + DIMENSION.key() + "=" + dimension);
+ }
+ List<Float> boxed = new ArrayList<>(vector.length);
+ for (float v : vector) {
+ boxed.add(v);
+ }
+ batchVectors[pending.textRowIndexes.get(i)] = boxed;
+ }
+ }
+
+ private void submitUpToWindow() {
+ while (inflight.size() < maxInflight && inputItr.hasNext()) {
+ Row[] rows = new Row[batchSize];
+ int count = 0;
+ List<String> texts = new ArrayList<>(batchSize);
+ List<Integer> textRowIndexes = new ArrayList<>(batchSize);
+ while (inputItr.hasNext() && count < batchSize) {
+ Row row = inputItr.next();
+ rows[count] = row;
+ String text = row.isNullAt(sourceIndex) ? null :
row.getString(sourceIndex);
+ if (text != null && !text.trim().isEmpty()) {
+ texts.add(text.length() > inputMaxChars ? text.substring(0,
inputMaxChars) : text);
+ textRowIndexes.add(count);
+ }
+ count++;
+ }
+ Future<List<float[]>> vectors = texts.isEmpty()
+ ?
CompletableFuture.completedFuture(java.util.Collections.<float[]>emptyList())
+ : executor().submit(() -> embed(texts));
+ inflight.add(new PendingBatch(rows, count, textRowIndexes, vectors));
+ }
+ }
+
+ private ExecutorService executor() {
+ if (executor == null) {
+ executor = Executors.newFixedThreadPool(maxInflight,
+ new CustomizedThreadFactory("embedding-transformer", true));
+ }
+ return executor;
+ }
+
+ private List<float[]> embed(List<String> texts) {
+ return providerInstance().embed(texts);
+ }
+
+ // called from the worker pool threads; synchronized so exactly one
provider is built
+ private synchronized EmbeddingProvider providerInstance() {
+ if (provider == null) {
+ EmbeddingProvider loaded = (EmbeddingProvider)
ReflectionUtils.loadClass(providerClass);
+ loaded.init(props);
+ provider = loaded;
+ }
+ return provider;
+ }
+ }
+}
diff --git
a/hudi-utilities/src/main/java/org/apache/hudi/utilities/transform/embedding/OpenAICompatibleEmbeddingProvider.java
b/hudi-utilities/src/main/java/org/apache/hudi/utilities/transform/embedding/OpenAICompatibleEmbeddingProvider.java
new file mode 100644
index 000000000000..3c3de75d2993
--- /dev/null
+++
b/hudi-utilities/src/main/java/org/apache/hudi/utilities/transform/embedding/OpenAICompatibleEmbeddingProvider.java
@@ -0,0 +1,180 @@
+/*
+ * 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.utilities.transform.embedding;
+
+import org.apache.hudi.common.config.TypedProperties;
+import org.apache.hudi.exception.HoodieException;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ArrayNode;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+
+import java.io.IOException;
+import java.net.URI;
+import java.net.http.HttpClient;
+import java.net.http.HttpRequest;
+import java.net.http.HttpResponse;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.apache.hudi.common.util.ConfigUtils.getLongWithAltKeys;
+import static org.apache.hudi.common.util.ConfigUtils.getStringWithAltKeys;
+import static
org.apache.hudi.utilities.config.EmbeddingTransformerConfig.API_KEY_ENV;
+import static
org.apache.hudi.utilities.config.EmbeddingTransformerConfig.ENDPOINT_URL;
+import static
org.apache.hudi.utilities.config.EmbeddingTransformerConfig.MODEL;
+import static
org.apache.hudi.utilities.config.EmbeddingTransformerConfig.TIMEOUT_MS;
+
+/**
+ * {@link EmbeddingProvider} for any OpenAI-compatible {@code /v1/embeddings}
endpoint
+ * (Ollama, TEI, vLLM, OpenAI, Voyage). One POST per record batch; transient
failures
+ * (429 and 5xx) are retried with exponential backoff honoring {@code
Retry-After}.
+ */
+public class OpenAICompatibleEmbeddingProvider implements EmbeddingProvider {
+
+ private static final long serialVersionUID = 1L;
+ private static final ObjectMapper MAPPER = new ObjectMapper();
+ private static final int MAX_ATTEMPTS = 5;
+ private static final long BASE_BACKOFF_MS = 1_000L;
+
+ private String endpointUrl;
+ private String model;
+ private String apiKeyEnv;
+ private long timeoutMs;
+
+ private transient HttpClient client;
+
+ @Override
+ public void init(TypedProperties props) {
+ this.endpointUrl = getStringWithAltKeys(props, ENDPOINT_URL);
+ this.model = getStringWithAltKeys(props, MODEL);
+ this.apiKeyEnv = getStringWithAltKeys(props, API_KEY_ENV, true);
+ this.timeoutMs = getLongWithAltKeys(props, TIMEOUT_MS);
+ }
+
+ @Override
+ public List<float[]> embed(List<String> texts) {
+ HttpRequest request = buildRequest(texts);
+ for (int attempt = 1; ; attempt++) {
+ try {
+ HttpResponse<String> response = client().send(request,
HttpResponse.BodyHandlers.ofString());
+ int status = response.statusCode();
+ if (status == 200) {
+ return parseVectors(response.body(), texts.size());
+ }
+ boolean retriable = status == 429 || status >= 500;
+ if (!retriable || attempt == MAX_ATTEMPTS) {
+ throw new HoodieException("Embeddings API returned HTTP " + status
+ + " (attempt " + attempt + "/" + MAX_ATTEMPTS + "): " +
truncate(response.body()));
+ }
+ sleep(backoffMs(response, attempt));
+ } catch (IOException e) {
+ if (attempt == MAX_ATTEMPTS) {
+ throw new HoodieException("Embeddings API unreachable after " +
MAX_ATTEMPTS + " attempts", e);
+ }
+ sleep(backoffMs(null, attempt));
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new HoodieException("Interrupted while calling embeddings API",
e);
+ }
+ }
+ }
+
+ // one client per provider instance, shared by concurrent batch requests:
HttpClient is
+ // thread-safe and keep-alives/pools connections internally, so requests
reuse sockets
+ private synchronized HttpClient client() {
+ if (client == null) {
+ client =
HttpClient.newBuilder().connectTimeout(Duration.ofMillis(timeoutMs)).build();
+ }
+ return client;
+ }
+
+ private HttpRequest buildRequest(List<String> texts) {
+ ObjectNode body = MAPPER.createObjectNode();
+ body.put("model", model);
+ ArrayNode input = body.putArray("input");
+ texts.forEach(input::add);
+ HttpRequest.Builder builder = HttpRequest.newBuilder()
+ .uri(URI.create(endpointUrl))
+ .timeout(Duration.ofMillis(timeoutMs))
+ .header("Content-Type", "application/json")
+ .POST(HttpRequest.BodyPublishers.ofString(body.toString()));
+ if (apiKeyEnv != null && !apiKeyEnv.isEmpty()) {
+ String apiKey = System.getenv(apiKeyEnv);
+ if (apiKey == null || apiKey.isEmpty()) {
+ throw new HoodieException("Environment variable " + apiKeyEnv + "
(from "
+ + API_KEY_ENV.key() + ") is not set");
+ }
+ builder.header("Authorization", "Bearer " + apiKey);
+ }
+ return builder.build();
+ }
+
+ private List<float[]> parseVectors(String responseBody, int expectedCount)
throws IOException {
+ JsonNode data = MAPPER.readTree(responseBody).path("data");
+ if (!data.isArray() || data.size() != expectedCount) {
+ throw new HoodieException("Embeddings API returned " + data.size()
+ + " vectors for " + expectedCount + " inputs");
+ }
+ List<float[]> vectors = new ArrayList<>(expectedCount);
+ // OpenAI-compatible APIs return entries ordered by `index`
+ for (JsonNode entry : data) {
+ JsonNode embedding = entry.path("embedding");
+ float[] vector = new float[embedding.size()];
+ for (int i = 0; i < vector.length; i++) {
+ vector[i] = (float) embedding.get(i).asDouble();
+ }
+ vectors.add(vector);
+ }
+ return vectors;
+ }
+
+ private static long backoffMs(HttpResponse<String> response, int attempt) {
+ if (response != null) {
+ // honor Retry-After (seconds form) when the server provides it
+ long retryAfter = response.headers().firstValue("Retry-After")
+ .map(v -> {
+ try {
+ return Long.parseLong(v.trim()) * 1000L;
+ } catch (NumberFormatException e) {
+ return -1L;
+ }
+ }).orElse(-1L);
+ if (retryAfter > 0) {
+ return retryAfter;
+ }
+ }
+ return BASE_BACKOFF_MS * (1L << (attempt - 1));
+ }
+
+ private static void sleep(long ms) {
+ try {
+ Thread.sleep(ms);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new HoodieException("Interrupted during embeddings API backoff",
e);
+ }
+ }
+
+ private static String truncate(String body) {
+ return body == null ? "" : body.substring(0, Math.min(body.length(), 500));
+ }
+}
diff --git
a/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/TestUnstructuredFileDFSSource.java
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/TestUnstructuredFileDFSSource.java
new file mode 100644
index 000000000000..08af5191fd6f
--- /dev/null
+++
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/TestUnstructuredFileDFSSource.java
@@ -0,0 +1,138 @@
+/*
+ * 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.utilities.sources;
+
+import org.apache.hudi.common.config.TypedProperties;
+import org.apache.hudi.common.schema.HoodieSchema;
+import org.apache.hudi.common.schema.HoodieSchemaType;
+import org.apache.hudi.common.table.checkpoint.Checkpoint;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.common.util.collection.Pair;
+import org.apache.hudi.utilities.config.UnstructuredFileSourceConfig;
+import org.apache.hudi.utilities.testutils.UtilitiesTestBase;
+
+import org.apache.avro.LogicalType;
+import org.apache.avro.Schema;
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Row;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.Arrays;
+import java.util.List;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Verifies the unstructured DFS source end to end at the source level:
per-record
+ * inline vs out-of-line blob placement, parse/chunk columns, BLOB schema
metadata
+ * propagation to Avro, and checkpoint advancement.
+ */
+public class TestUnstructuredFileDFSSource extends UtilitiesTestBase {
+
+ @TempDir
+ static Path tempDir;
+
+ @BeforeAll
+ public static void setupOnce() throws Exception {
+ initTestServices();
+ }
+
+ private TypedProperties props(long inlineMaxBytes) {
+ TypedProperties props = new TypedProperties();
+ props.setProperty("hoodie.streamer.source.dfs.root", tempDir.toString());
+
props.setProperty(UnstructuredFileSourceConfig.BLOB_INLINE_MAX_BYTES.key(),
String.valueOf(inlineMaxBytes));
+ return props;
+ }
+
+ @Test
+ public void testInlineVsOutOfLineParseColumnsAndCheckpoint() throws
IOException {
+ Files.write(tempDir.resolve("small.txt"),
+ "hudi unstructured ingest smoke
text".getBytes(StandardCharsets.UTF_8));
+ byte[] big = new byte[256];
+ Arrays.fill(big, (byte) 'x');
+ Files.write(tempDir.resolve("big.txt"), big);
+
+ // threshold between the two files: 35-byte file inlines, 256-byte file
goes out-of-line
+ UnstructuredFileDFSSource source =
+ new UnstructuredFileDFSSource(props(100), jsc, sparkSession, null);
+ Pair<Option<Dataset<Row>>, Checkpoint> batch =
source.fetchNextBatch(Option.empty(), Long.MAX_VALUE);
+ assertTrue(batch.getLeft().isPresent());
+ Dataset<Row> df = batch.getLeft().get();
+ List<Row> rows = df.collectAsList();
+ assertEquals(2, rows.size());
+
+ Row small = rows.stream().filter(r ->
r.getString(df.schema().fieldIndex("file_name"))
+ .equals("small.txt")).findFirst().get();
+ Row smallBlob = small.getStruct(df.schema().fieldIndex("content"));
+ assertEquals(HoodieSchema.Blob.INLINE, smallBlob.getString(0));
+ assertEquals(35, ((byte[]) smallBlob.get(1)).length);
+ assertNull(smallBlob.get(2));
+ assertEquals("SUCCESS",
small.getString(df.schema().fieldIndex("parse_status")));
+
assertTrue(small.getString(df.schema().fieldIndex("extracted_text")).contains("smoke"));
+ assertFalse(small.getList(df.schema().fieldIndex("chunks")).isEmpty());
+ assertEquals("txt", small.getString(df.schema().fieldIndex("extension")));
+
+ Row bigRow = rows.stream().filter(r ->
r.getString(df.schema().fieldIndex("file_name"))
+ .equals("big.txt")).findFirst().get();
+ Row bigBlob = bigRow.getStruct(df.schema().fieldIndex("content"));
+ assertEquals(HoodieSchema.Blob.OUT_OF_LINE, bigBlob.getString(0));
+ assertNull(bigBlob.get(1));
+ Row reference = bigBlob.getStruct(2);
+ assertTrue(reference.getString(0).endsWith("big.txt"));
+ assertFalse(reference.getBoolean(3)); // managed=false: points at the
original file in place
+ // out-of-line files are still parsed (streamed from the source file)
+ assertEquals("SUCCESS",
bigRow.getString(df.schema().fieldIndex("parse_status")));
+
+ // checkpoint advanced; an immediate re-fetch returns empty
+ Checkpoint checkpoint = batch.getRight();
+ assertNotNull(checkpoint);
+ Pair<Option<Dataset<Row>>, Checkpoint> next =
source.fetchNextBatch(Option.of(checkpoint), Long.MAX_VALUE);
+ assertFalse(next.getLeft().isPresent());
+ }
+
+ @Test
+ public void testBlobLogicalTypeSurvivesAvroConversion() {
+ // The schema seam the whole design rests on: the BLOB metadata on the
source's
+ // StructType must convert into the Avro/Hoodie blob logical type.
+ HoodieSchema hoodieSchema = org.apache.hudi.HoodieSchemaConversionUtils
+
.convertStructTypeToHoodieSchema(UnstructuredFileDFSSource.SOURCE_SCHEMA,
"hoodie_source", "hoodie.source");
+ Schema avro = hoodieSchema.toAvroSchema();
+ Schema content = resolveNullable(avro.getField("content").schema());
+ LogicalType logicalType = content.getLogicalType();
+ assertNotNull(logicalType, "content field lost the blob logical type");
+ assertEquals(HoodieSchemaType.BLOB.name().toLowerCase(),
logicalType.getName().toLowerCase());
+ }
+
+ private static Schema resolveNullable(Schema schema) {
+ if (schema.getType() == Schema.Type.UNION) {
+ return schema.getTypes().stream().filter(s -> s.getType() !=
Schema.Type.NULL).findFirst().get();
+ }
+ return schema;
+ }
+}
diff --git
a/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/helpers/unstructured/TestTextChunker.java
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/helpers/unstructured/TestTextChunker.java
new file mode 100644
index 000000000000..2c88b55d5580
--- /dev/null
+++
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/helpers/unstructured/TestTextChunker.java
@@ -0,0 +1,96 @@
+/*
+ * 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.utilities.sources.helpers.unstructured;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.List;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class TestTextChunker {
+
+ @Test
+ public void testChunkBoundariesOverlapAndEdgeCases() {
+ TextChunker chunker = new TextChunker(10, 3);
+
+ // 22 chars, step 7: chunks start at 0, 7, 14; the third reaches the end
+ List<TextChunker.Chunk> chunks = chunker.chunk("abcdefghijklmnopqrstuv");
+ assertEquals(3, chunks.size());
+ assertEquals("abcdefghij", chunks.get(0).text);
+ assertEquals(0, chunks.get(0).charStart);
+ assertEquals("hijklmnopq", chunks.get(1).text); // overlaps previous by 3
+ assertEquals(7, chunks.get(1).charStart);
+ assertEquals("opqrstuv", chunks.get(2).text); // final partial chunk
+ assertEquals(2, chunks.get(2).chunkId);
+ assertEquals(14, chunks.get(2).charStart);
+
+ // text shorter than one chunk -> single chunk, exact text
+ List<TextChunker.Chunk> single = chunker.chunk("short");
+ assertEquals(1, single.size());
+ assertEquals("short", single.get(0).text);
+
+ // empty / null -> no chunks
+ assertTrue(chunker.chunk("").isEmpty());
+ assertTrue(chunker.chunk(null).isEmpty());
+
+ // invalid configs rejected
+ assertThrows(IllegalArgumentException.class, () -> new TextChunker(0, 0));
+ assertThrows(IllegalArgumentException.class, () -> new TextChunker(10,
10));
+ }
+
+ @Test
+ public void testBreaksAtNaturalBoundariesInPreferenceOrder() {
+ // paragraph break inside the window wins over the later word boundary
+ TextChunker chunker = new TextChunker(30, 0);
+ List<TextChunker.Chunk> paragraphs = chunker.chunk("First paragraph
here.\n\nSecond paragraph runs longer than one window");
+ assertEquals("First paragraph here.\n\n", paragraphs.get(0).text);
+ assertEquals(23, paragraphs.get(1).charStart);
+
+ // no paragraph/newline -> sentence end preferred over word boundary
+ List<TextChunker.Chunk> sentences = chunker.chunk("A short sentence ends.
The next sentence continues past the window");
+ assertEquals("A short sentence ends. ", sentences.get(0).text);
+
+ // no sentence end -> last word boundary inside the window
+ List<TextChunker.Chunk> words = chunker.chunk("words without punctuation
keep flowing past the window edge");
+ assertTrue(words.get(0).text.endsWith(" "));
+ assertTrue(words.get(0).text.length() <= 30);
+
+ // every chunk is a verbatim substring at its recorded offset
+ String text = "Mixed content.\n\nWith breaks. And words flowing on and on
beyond several windows of text";
+ for (TextChunker.Chunk c : chunker.chunk(text)) {
+ assertEquals(text.substring(c.charStart, c.charStart + c.text.length()),
c.text);
+ }
+ }
+
+ @Test
+ public void testBoundaryFloorGuaranteesProgressWithLargeOverlap() {
+ // overlap close to chunk size: boundary breaks must still advance each
chunk
+ TextChunker chunker = new TextChunker(10, 8);
+ String text = "a b c d e f g h i j k l m n o p q r s t";
+ List<TextChunker.Chunk> chunks = chunker.chunk(text);
+ for (int i = 1; i < chunks.size(); i++) {
+ assertTrue(chunks.get(i).charStart > chunks.get(i - 1).charStart);
+ }
+ assertEquals(text.length(), chunks.get(chunks.size() - 1).charStart
+ + chunks.get(chunks.size() - 1).text.length());
+ }
+}
diff --git
a/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/helpers/unstructured/TestTikaDocumentParser.java
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/helpers/unstructured/TestTikaDocumentParser.java
new file mode 100644
index 000000000000..edc03348b5bb
--- /dev/null
+++
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/helpers/unstructured/TestTikaDocumentParser.java
@@ -0,0 +1,172 @@
+/*
+ * 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.utilities.sources.helpers.unstructured;
+
+import org.junit.jupiter.api.Test;
+
+import java.io.ByteArrayInputStream;
+import java.nio.charset.StandardCharsets;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Verifies the never-throw contract and status mapping of the Tika-backed
parser.
+ * Fixtures are generated in-line (no binary files in the repo).
+ */
+public class TestTikaDocumentParser {
+
+ private final TikaDocumentParser parser = new TikaDocumentParser();
+
+ private ParseResult parse(byte[] content, String fileName, int maxChars) {
+ return parser.parse(new ByteArrayInputStream(content), fileName, maxChars);
+ }
+
+ @Test
+ public void testPlainTextAndHtmlSuccessWithMetadata() {
+ ParseResult text = parse("hudi ingests unstructured
data".getBytes(StandardCharsets.UTF_8),
+ "note.txt", 1000);
+ assertEquals(ParseResult.ParseStatus.SUCCESS, text.getStatus());
+ assertTrue(text.getText().contains("unstructured"));
+ assertFalse(text.getMetadata().isEmpty()); // content-type at minimum
+
+ ParseResult html = parse("<html><title>t</title><body><p>lakehouse
body</p></body></html>"
+ .getBytes(StandardCharsets.UTF_8), "page.html", 1000);
+ assertEquals(ParseResult.ParseStatus.SUCCESS, html.getStatus());
+ assertTrue(html.getText().contains("lakehouse body"));
+ assertFalse(html.getText().contains("<p>")); // markup stripped
+ }
+
+ @Test
+ public void testTruncationAtCharCap() {
+ byte[] longText = new byte[5000];
+ java.util.Arrays.fill(longText, (byte) 'a');
+ ParseResult result = parse(longText, "big.txt", 100);
+ assertEquals(ParseResult.ParseStatus.TRUNCATED, result.getStatus());
+ assertTrue(result.getText().length() <= 100 + 1);
+ }
+
+ @Test
+ public void testPdfWithTextLayerExtractsTextAndMetadata() {
+ ParseResult pdf = parse(minimalPdf("Espresso extraction physics for
lakehouses"),
+ "doc.pdf", 1000);
+ assertEquals(ParseResult.ParseStatus.SUCCESS, pdf.getStatus());
+ assertTrue(pdf.getText().contains("Espresso extraction physics"));
+ assertTrue(pdf.getMetadata().getOrDefault("Content-Type",
"").contains("pdf"));
+ }
+
+ @Test
+ public void testDocxExtractsText() {
+ ParseResult docx = parse(minimalDocx("Coral reef ecology field notes"),
"doc.docx", 1000);
+ assertEquals(ParseResult.ParseStatus.SUCCESS, docx.getStatus());
+ assertTrue(docx.getText().contains("Coral reef ecology"));
+ }
+
+ @Test
+ public void testMarkdownAndCsvParseAsText() {
+ ParseResult md = parse("# Heading\n\nBody with
*emphasis*".getBytes(StandardCharsets.UTF_8),
+ "notes.md", 1000);
+ assertEquals(ParseResult.ParseStatus.SUCCESS, md.getStatus());
+ assertTrue(md.getText().contains("Body with"));
+
+ ParseResult csv =
parse("city,count\nparis,2".getBytes(StandardCharsets.UTF_8),
+ "table.csv", 1000);
+ assertEquals(ParseResult.ParseStatus.SUCCESS, csv.getStatus());
+ assertTrue(csv.getText().contains("paris"));
+ }
+
+ @Test
+ public void testCorruptAndBinaryInputsNeverThrow() {
+ // Claims to be PDF (magic bytes) but is corrupt: parser error -> FAILED,
not thrown
+ ParseResult corrupt = parse("%PDF-1.4 this is not really a
pdf".getBytes(StandardCharsets.UTF_8),
+ "corrupt.pdf", 1000);
+ assertEquals(ParseResult.ParseStatus.FAILED, corrupt.getStatus());
+ assertNotNull(corrupt.getError());
+
+ // Unrecognizable binary: no text, no error -> EMPTY
+ ParseResult binary = parse(new byte[] {0x00, 0x11, 0x22, 0x33, (byte)
0xff}, "blob.bin", 1000);
+ assertEquals(ParseResult.ParseStatus.EMPTY, binary.getStatus());
+ assertEquals("", binary.getText());
+ }
+
+ /**
+ * Smallest well-formed single-page PDF with a real text layer.
+ */
+ private static byte[] minimalPdf(String text) {
+ byte[] stream = ("BT /F1 12 Tf 50 750 Td (" + text + ") Tj
ET").getBytes(StandardCharsets.UTF_8);
+ String[] objects = {
+ "<< /Type /Catalog /Pages 2 0 R >>",
+ "<< /Type /Pages /Kids [3 0 R] /Count 1 >>",
+ "<< /Type /Page /Parent 2 0 R /MediaBox [0 0 612 792] /Contents 4 0 R "
+ + "/Resources << /Font << /F1 5 0 R >> >> >>",
+ "<< /Length " + stream.length + " >>\nstream\n" + new String(stream,
StandardCharsets.UTF_8)
+ + "\nendstream",
+ "<< /Type /Font /Subtype /Type1 /BaseFont /Helvetica >>"};
+ StringBuilder pdf = new StringBuilder("%PDF-1.4\n");
+ int[] offsets = new int[objects.length];
+ for (int i = 0; i < objects.length; i++) {
+ offsets[i] = pdf.length();
+ pdf.append(i + 1).append(" 0
obj\n").append(objects[i]).append("\nendobj\n");
+ }
+ int xref = pdf.length();
+ pdf.append("xref\n0 ").append(objects.length + 1).append("\n0000000000
65535 f \n");
+ for (int offset : offsets) {
+ pdf.append(String.format("%010d 00000 n \n", offset));
+ }
+ pdf.append("trailer\n<< /Size ").append(objects.length + 1)
+ .append(" /Root 1 0 R >>\nstartxref\n").append(xref).append("\n%%EOF");
+ return pdf.toString().getBytes(StandardCharsets.UTF_8);
+ }
+
+ /**
+ * Smallest well-formed DOCX (OOXML zip with content types, relationships,
one paragraph).
+ */
+ private static byte[] minimalDocx(String text) {
+ String contentTypes = "<?xml version=\"1.0\" encoding=\"UTF-8\"
standalone=\"yes\"?>"
+ + "<Types
xmlns=\"http://schemas.openxmlformats.org/package/2006/content-types\";>"
+ + "<Default Extension=\"rels\"
ContentType=\"application/vnd.openxmlformats-package.relationships+xml\"/>"
+ + "<Default Extension=\"xml\" ContentType=\"application/xml\"/>"
+ + "<Override PartName=\"/word/document.xml\" "
+ +
"ContentType=\"application/vnd.openxmlformats-officedocument.wordprocessingml.document.main+xml\"/></Types>";
+ String rels = "<?xml version=\"1.0\" encoding=\"UTF-8\"
standalone=\"yes\"?>"
+ + "<Relationships
xmlns=\"http://schemas.openxmlformats.org/package/2006/relationships\";>"
+ + "<Relationship Id=\"rId1\" "
+ +
"Type=\"http://schemas.openxmlformats.org/officeDocument/2006/relationships/officeDocument\";
"
+ + "Target=\"word/document.xml\"/></Relationships>";
+ String document = "<?xml version=\"1.0\" encoding=\"UTF-8\"
standalone=\"yes\"?>"
+ + "<w:document
xmlns:w=\"http://schemas.openxmlformats.org/wordprocessingml/2006/main\";>"
+ + "<w:body><w:p><w:r><w:t>" + text +
"</w:t></w:r></w:p></w:body></w:document>";
+ try {
+ java.io.ByteArrayOutputStream buffer = new
java.io.ByteArrayOutputStream();
+ try (java.util.zip.ZipOutputStream zip = new
java.util.zip.ZipOutputStream(buffer)) {
+ for (String[] entry : new String[][] {
+ {"[Content_Types].xml", contentTypes}, {"_rels/.rels", rels},
{"word/document.xml", document}}) {
+ zip.putNextEntry(new java.util.zip.ZipEntry(entry[0]));
+ zip.write(entry[1].getBytes(StandardCharsets.UTF_8));
+ zip.closeEntry();
+ }
+ }
+ return buffer.toByteArray();
+ } catch (java.io.IOException e) {
+ throw new RuntimeException(e);
+ }
+ }
+}
diff --git
a/hudi-utilities/src/test/java/org/apache/hudi/utilities/streamer/TestUnstructuredIngestE2E.java
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/streamer/TestUnstructuredIngestE2E.java
new file mode 100644
index 000000000000..6174ff0243e6
--- /dev/null
+++
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/streamer/TestUnstructuredIngestE2E.java
@@ -0,0 +1,198 @@
+/*
+ * 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.utilities.streamer;
+
+import org.apache.hudi.common.config.TypedProperties;
+import org.apache.hudi.common.model.WriteOperationType;
+import org.apache.hudi.common.schema.HoodieSchema;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.table.TableSchemaResolver;
+import org.apache.hudi.common.testutils.HoodieTestUtils;
+import org.apache.hudi.utilities.config.EmbeddingTransformerConfig;
+import org.apache.hudi.utilities.config.UnstructuredFileSourceConfig;
+import org.apache.hudi.utilities.deltastreamer.HoodieDeltaStreamer;
+import org.apache.hudi.utilities.deltastreamer.HoodieDeltaStreamerTestBase;
+import org.apache.hudi.utilities.sources.UnstructuredFileDFSSource;
+import org.apache.hudi.utilities.transform.embedding.EmbeddingTransformer;
+
+import com.sun.net.httpserver.HttpServer;
+import org.apache.avro.Schema;
+import org.apache.hadoop.fs.FSDataOutputStream;
+import org.apache.hadoop.fs.Path;
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Row;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+import java.io.OutputStream;
+import java.net.InetSocketAddress;
+import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Streamer-level round trip for the unstructured file source on a PARQUET COW
table:
+ * one sync ingesting a small (INLINE) and a large (OUT_OF_LINE) file, blob
struct and
+ * parse columns verified through the Hudi read path, then a second sync after
modifying
+ * the small file, verifying upsert-by-path keeps the table current with the
source dir.
+ */
+public class TestUnstructuredIngestE2E extends HoodieDeltaStreamerTestBase {
+
+ private static final String PROPS_FILE =
"test-unstructured-source.properties";
+ private static final int INLINE_THRESHOLD = 1024;
+ private static final int DIM = 2;
+ private static final Pattern INPUT_PATTERN =
Pattern.compile("\"input\":\\[(.*?)\\]");
+
+ private static HttpServer embeddingStub;
+
+ @BeforeAll
+ public static void startEmbeddingStub() throws IOException {
+ embeddingStub = HttpServer.create(new InetSocketAddress(0), 0);
+ embeddingStub.createContext("/v1/embeddings", exchange -> {
+ java.io.ByteArrayOutputStream buffer = new
java.io.ByteArrayOutputStream();
+ byte[] chunk = new byte[4096];
+ int n;
+ while ((n = exchange.getRequestBody().read(chunk)) > 0) {
+ buffer.write(chunk, 0, n);
+ }
+ String body = new String(buffer.toByteArray(), StandardCharsets.UTF_8);
+ Matcher matcher = INPUT_PATTERN.matcher(body);
+ StringBuilder data = new StringBuilder("{\"data\":[");
+ if (matcher.find()) {
+ String[] inputs = matcher.group(1).split("\",\"");
+ for (int i = 0; i < inputs.length; i++) {
+ // deterministic: first component = input text length
+ data.append(i > 0 ? "," : "").append("{\"index\":").append(i)
+
.append(",\"embedding\":[").append(inputs[i].replaceAll("^\"|\"$", "").length())
+ .append(".0,0.5]}");
+ }
+ }
+ data.append("]}");
+ byte[] response = data.toString().getBytes(StandardCharsets.UTF_8);
+ exchange.getResponseHeaders().add("Content-Type", "application/json");
+ exchange.sendResponseHeaders(200, response.length);
+ try (OutputStream out = exchange.getResponseBody()) {
+ out.write(response);
+ }
+ });
+ embeddingStub.start();
+ }
+
+ @AfterAll
+ public static void stopEmbeddingStub() {
+ if (embeddingStub != null) {
+ embeddingStub.stop(0);
+ }
+ }
+
+ private void writeSourceFile(String dir, String name, byte[] content) throws
IOException {
+ try (FSDataOutputStream out = fs.create(new Path(dir, name), true)) {
+ out.write(content);
+ }
+ }
+
+ @Test
+ public void testIngestAndUpsertBlobTable() throws Exception {
+ String sourceRoot = basePath + "/unstructured-input";
+ String tableBasePath = basePath + "/unstructured_table";
+ writeSourceFile(sourceRoot, "small.txt", "hudi lakehouse smoke
document".getBytes(StandardCharsets.UTF_8));
+ byte[] big = new byte[INLINE_THRESHOLD * 4];
+ Arrays.fill(big, (byte) 'b');
+ writeSourceFile(sourceRoot, "big.txt", big);
+
+ TypedProperties props = new TypedProperties();
+ props.setProperty("hoodie.streamer.source.dfs.root", sourceRoot);
+
props.setProperty(UnstructuredFileSourceConfig.BLOB_INLINE_MAX_BYTES.key(),
String.valueOf(INLINE_THRESHOLD));
+ props.setProperty("hoodie.datasource.write.recordkey.field", "path");
+ props.setProperty("hoodie.datasource.write.partitionpath.field",
"extension");
+ props.setProperty(EmbeddingTransformerConfig.ENDPOINT_URL.key(),
+ "http://localhost:"; + embeddingStub.getAddress().getPort() +
"/v1/embeddings");
+ props.setProperty(EmbeddingTransformerConfig.MODEL.key(), "stub-model");
+ props.setProperty(EmbeddingTransformerConfig.DIMENSION.key(),
String.valueOf(DIM));
+ Helpers.savePropsToDFS(props, storage, basePath + "/" + PROPS_FILE);
+
+ HoodieDeltaStreamer.Config cfg = TestHelpers.makeConfig(tableBasePath,
WriteOperationType.UPSERT,
+ UnstructuredFileDFSSource.class.getName(),
+ Collections.singletonList(EmbeddingTransformer.class.getName()),
PROPS_FILE, false, false,
+ 100_000_000, false, null, "COPY_ON_WRITE", "modification_time", null);
+ new HoodieDeltaStreamer(cfg, jsc).sync();
+
+ Dataset<Row> table =
sparkSession.read().format("hudi").load(tableBasePath);
+ List<Row> rows = table.collectAsList();
+ assertEquals(2, rows.size());
+
+ Row small = rows.stream().filter(r ->
r.getString(table.schema().fieldIndex("file_name"))
+ .equals("small.txt")).findFirst().get();
+ Row smallBlob = small.getStruct(table.schema().fieldIndex("content"));
+ assertEquals(HoodieSchema.Blob.INLINE, smallBlob.getString(0));
+ assertEquals("hudi lakehouse smoke document",
+ new String((byte[]) smallBlob.get(1), StandardCharsets.UTF_8));
+ assertEquals("SUCCESS",
small.getString(table.schema().fieldIndex("parse_status")));
+
assertTrue(small.getString(table.schema().fieldIndex("extracted_text")).contains("smoke"));
+
+ Row bigRow = rows.stream().filter(r ->
r.getString(table.schema().fieldIndex("file_name"))
+ .equals("big.txt")).findFirst().get();
+ Row bigBlob = bigRow.getStruct(table.schema().fieldIndex("content"));
+ assertEquals(HoodieSchema.Blob.OUT_OF_LINE, bigBlob.getString(0));
+ assertNull(bigBlob.get(1));
+ assertTrue(bigBlob.getStruct(2).getString(0).endsWith("big.txt"));
+
+ // embeddings populated by the transformer via the stub API (value = text
length)
+ int embeddingIndex = table.schema().fieldIndex("embedding");
+ List<Float> smallVector = small.getList(embeddingIndex);
+ assertEquals(DIM, smallVector.size());
+ assertEquals((float) "hudi lakehouse smoke document".length(),
smallVector.get(0));
+
+ // the committed table schema carries the vector logical type end to end
+ HoodieTableMetaClient metaClient =
HoodieTestUtils.createMetaClient(storage, tableBasePath);
+ Schema tableSchema = new
TableSchemaResolver(metaClient).getTableSchema(false).toAvroSchema();
+ Schema embeddingField = tableSchema.getField("embedding").schema();
+ Schema embeddingType = embeddingField.getType() == Schema.Type.UNION
+ ? embeddingField.getTypes().stream().filter(t -> t.getType() !=
Schema.Type.NULL).findFirst().get()
+ : embeddingField;
+ assertNotNull(embeddingType.getLogicalType(), "embedding column lost the
vector logical type");
+
assertTrue(embeddingType.getLogicalType().getName().toLowerCase().contains("vector"));
+
+ // Second sync after the source file changes: upsert-by-path refreshes the
row in place.
+ writeSourceFile(sourceRoot, "small.txt",
+ "hudi lakehouse refreshed document".getBytes(StandardCharsets.UTF_8));
+ new HoodieDeltaStreamer(cfg, jsc).sync();
+
+ Dataset<Row> refreshed =
sparkSession.read().format("hudi").load(tableBasePath);
+ assertEquals(2, refreshed.count());
+ Row refreshedSmall = refreshed.collectAsList().stream()
+ .filter(r ->
r.getString(refreshed.schema().fieldIndex("file_name")).equals("small.txt"))
+ .findFirst().get();
+
assertTrue(refreshedSmall.getString(refreshed.schema().fieldIndex("extracted_text"))
+ .contains("refreshed"));
+ // the embedding refreshed with the text (stub vector tracks text length)
+ assertEquals((float) "hudi lakehouse refreshed document".length(),
+
refreshedSmall.<Float>getList(refreshed.schema().fieldIndex("embedding")).get(0));
+ }
+}
diff --git
a/hudi-utilities/src/test/java/org/apache/hudi/utilities/transform/embedding/TestEmbeddingTransformer.java
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/transform/embedding/TestEmbeddingTransformer.java
new file mode 100644
index 000000000000..773fbd7494a0
--- /dev/null
+++
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/transform/embedding/TestEmbeddingTransformer.java
@@ -0,0 +1,201 @@
+/*
+ * 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.utilities.transform.embedding;
+
+import org.apache.hudi.common.config.TypedProperties;
+import org.apache.hudi.common.schema.HoodieSchema;
+import org.apache.hudi.utilities.config.EmbeddingTransformerConfig;
+import org.apache.hudi.utilities.testutils.UtilitiesTestBase;
+
+import com.sun.net.httpserver.HttpServer;
+import org.apache.spark.SparkException;
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Row;
+import org.apache.spark.sql.RowFactory;
+import org.apache.spark.sql.types.DataTypes;
+import org.apache.spark.sql.types.StructField;
+import org.apache.spark.sql.types.StructType;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+import java.io.OutputStream;
+import java.net.InetSocketAddress;
+import java.nio.charset.StandardCharsets;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Embedding transformer against a stub OpenAI-compatible server: record-level
+ * batching, deterministic vectors, null vectors for text-less rows, VECTOR
+ * metadata on both apply() and transformedSchema(), 429-retry, and hard
failure
+ * on persistent errors.
+ */
+public class TestEmbeddingTransformer extends UtilitiesTestBase {
+
+ private static final int DIM = 2;
+ private static final Pattern INPUT_PATTERN =
Pattern.compile("\"input\":\\[(.*?)\\]");
+
+ private static HttpServer server;
+ private static final AtomicInteger REQUEST_COUNT = new AtomicInteger();
+ private static final AtomicInteger REMAINING_FAILURES = new AtomicInteger();
+ private static volatile int failureStatus = 500;
+
+ @BeforeAll
+ public static void setupAll() throws Exception {
+ initTestServices();
+ server = HttpServer.create(new InetSocketAddress(0), 0);
+ server.createContext("/v1/embeddings", exchange -> {
+ REQUEST_COUNT.incrementAndGet();
+ if (REMAINING_FAILURES.getAndUpdate(n -> Math.max(0, n - 1)) > 0) {
+ exchange.getResponseHeaders().add("Retry-After", "1");
+ exchange.sendResponseHeaders(failureStatus, -1);
+ exchange.close();
+ return;
+ }
+ String body = new String(readAll(exchange.getRequestBody()),
StandardCharsets.UTF_8);
+ // deterministic stub: vector = [textLength, 0.5] per input, in request
order
+ StringBuilder data = new StringBuilder("{\"data\":[");
+ Matcher matcher = INPUT_PATTERN.matcher(body);
+ assertTrue(matcher.find());
+ String[] inputs = matcher.group(1).split("\",\"");
+ for (int i = 0; i < inputs.length; i++) {
+ String text = inputs[i].replaceAll("^\"|\"$", "");
+ data.append(i > 0 ? "," : "")
+ .append("{\"index\":").append(i)
+
.append(",\"embedding\":[").append(text.length()).append(".0,0.5]}");
+ }
+ data.append("]}");
+ byte[] response = data.toString().getBytes(StandardCharsets.UTF_8);
+ exchange.getResponseHeaders().add("Content-Type", "application/json");
+ exchange.sendResponseHeaders(200, response.length);
+ try (OutputStream out = exchange.getResponseBody()) {
+ out.write(response);
+ }
+ });
+ server.start();
+ }
+
+ @AfterAll
+ public static void teardownAll() {
+ if (server != null) {
+ server.stop(0);
+ }
+ }
+
+ private static byte[] readAll(java.io.InputStream in) throws IOException {
+ java.io.ByteArrayOutputStream buffer = new java.io.ByteArrayOutputStream();
+ byte[] chunk = new byte[4096];
+ int n;
+ while ((n = in.read(chunk)) > 0) {
+ buffer.write(chunk, 0, n);
+ }
+ return buffer.toByteArray();
+ }
+
+ private TypedProperties props(int batchSize) {
+ TypedProperties props = new TypedProperties();
+ props.setProperty(EmbeddingTransformerConfig.ENDPOINT_URL.key(),
+ "http://localhost:"; + server.getAddress().getPort() +
"/v1/embeddings");
+ props.setProperty(EmbeddingTransformerConfig.MODEL.key(), "stub-model");
+ props.setProperty(EmbeddingTransformerConfig.DIMENSION.key(),
String.valueOf(DIM));
+ props.setProperty(EmbeddingTransformerConfig.BATCH_SIZE.key(),
String.valueOf(batchSize));
+ return props;
+ }
+
+ private Dataset<Row> sourceDataset(String... texts) {
+ StructType schema = new StructType(new StructField[] {
+ DataTypes.createStructField("path", DataTypes.StringType, false),
+ DataTypes.createStructField("extracted_text", DataTypes.StringType,
true)});
+ List<Row> rows = new java.util.ArrayList<>();
+ for (int i = 0; i < texts.length; i++) {
+ rows.add(RowFactory.create("file-" + i, texts[i]));
+ }
+ return sparkSession.createDataFrame(rows, schema);
+ }
+
+ @Test
+ public void testBatchingVectorsAndNullForEmptyText() {
+ REQUEST_COUNT.set(0);
+ // 5 rows in one partition, 4 with text, batch size 2 -> rows buffer as
+ // [2 texts][2 texts incl. the empty-text row][1 text] = 3 API calls
+ Dataset<Row> input = sourceDataset("alpha", "bete", "", "gamma7",
"epsilon90").coalesce(1);
+ Dataset<Row> output = new EmbeddingTransformer().apply(jsc, sparkSession,
input, props(2));
+
+ List<Row> rows = output.collectAsList();
+ assertEquals(5, rows.size());
+ int vectorIndex = output.schema().fieldIndex("embedding");
+ assertEquals(3, REQUEST_COUNT.get());
+
+ for (Row row : rows) {
+ String text = row.getString(1);
+ if (text == null || text.isEmpty()) {
+ assertNull(row.get(vectorIndex)); // text-less rows are never sent to
the API
+ } else {
+ List<Float> vector = row.getList(vectorIndex);
+ assertEquals(DIM, vector.size());
+ assertEquals((float) text.length(), vector.get(0)); // deterministic
stub value
+ }
+ }
+ }
+
+ @Test
+ public void testVectorMetadataOnApplyAndTransformedSchema() {
+ Dataset<Row> input = sourceDataset("one");
+ EmbeddingTransformer transformer = new EmbeddingTransformer();
+
+ StructField applied = new StructType(
+ transformer.apply(jsc, sparkSession, input,
props(16)).schema().fields())
+ .apply("embedding");
+ assertEquals("VECTOR(" + DIM + ")",
+ applied.metadata().getString(HoodieSchema.TYPE_METADATA_FIELD));
+
+ StructField declared = transformer
+ .transformedSchema(jsc, sparkSession, input.schema(),
props(16)).apply("embedding");
+ assertEquals("VECTOR(" + DIM + ")",
+ declared.metadata().getString(HoodieSchema.TYPE_METADATA_FIELD));
+ }
+
+ @Test
+ public void testRetryOn429ThenSuccessAndFailFastOnClientError() {
+ // one 429 (with Retry-After) followed by success -> transform completes
+ REMAINING_FAILURES.set(1);
+ failureStatus = 429;
+ Dataset<Row> ok = new EmbeddingTransformer()
+ .apply(jsc, sparkSession, sourceDataset("retryable").coalesce(1),
props(16));
+ assertEquals(1, ok.collectAsList().size());
+
+ // non-retriable client error -> the batch fails loudly and immediately
+ // (no silent null vectors, no pointless backoff)
+ REMAINING_FAILURES.set(Integer.MAX_VALUE);
+ failureStatus = 400;
+ Dataset<Row> failing = new EmbeddingTransformer()
+ .apply(jsc, sparkSession, sourceDataset("doomed").coalesce(1),
props(16));
+ assertThrows(SparkException.class, failing::collectAsList);
+ REMAINING_FAILURES.set(0);
+ }
+}
diff --git
a/hudi-utilities/src/test/java/org/apache/hudi/utilities/transform/embedding/TestOpenAICompatibleEmbeddingProvider.java
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/transform/embedding/TestOpenAICompatibleEmbeddingProvider.java
new file mode 100644
index 000000000000..466738524796
--- /dev/null
+++
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/transform/embedding/TestOpenAICompatibleEmbeddingProvider.java
@@ -0,0 +1,181 @@
+/*
+ * 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.utilities.transform.embedding;
+
+import org.apache.hudi.common.config.TypedProperties;
+import org.apache.hudi.exception.HoodieException;
+import org.apache.hudi.utilities.config.EmbeddingTransformerConfig;
+
+import com.sun.net.httpserver.HttpServer;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.io.OutputStream;
+import java.net.InetSocketAddress;
+import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Direct unit tests for {@link OpenAICompatibleEmbeddingProvider} against a
stub
+ * server (no Spark): request shape and auth header, vector parsing, retry on
5xx
+ * honoring Retry-After, retry on connection-level IOException, retry
exhaustion,
+ * fail-fast on 4xx, and response/input count mismatch.
+ */
+public class TestOpenAICompatibleEmbeddingProvider {
+
+ private static HttpServer server;
+ private static final AtomicInteger REQUEST_COUNT = new AtomicInteger();
+ private static final AtomicInteger REMAINING_FAILURES = new AtomicInteger();
+ private static volatile int failureStatus = 500;
+ private static volatile boolean abortConnection = false;
+ private static volatile int vectorsToReturn = -1; // -1 = one per input
+ private static final AtomicReference<String> LAST_BODY = new
AtomicReference<>();
+ private static final AtomicReference<String> LAST_AUTH = new
AtomicReference<>();
+
+ @BeforeAll
+ public static void startServer() throws Exception {
+ server = HttpServer.create(new InetSocketAddress(0), 0);
+ server.createContext("/v1/embeddings", exchange -> {
+ REQUEST_COUNT.incrementAndGet();
+ LAST_AUTH.set(exchange.getRequestHeaders().getFirst("Authorization"));
+ byte[] request = new byte[8192];
+ int n = exchange.getRequestBody().read(request);
+ LAST_BODY.set(new String(request, 0, Math.max(0, n),
StandardCharsets.UTF_8));
+ if (REMAINING_FAILURES.getAndUpdate(f -> Math.max(0, f - 1)) > 0) {
+ if (abortConnection) {
+ exchange.close(); // no response -> client-side IOException
+ return;
+ }
+ exchange.getResponseHeaders().add("Retry-After", "1");
+ exchange.sendResponseHeaders(failureStatus, -1);
+ exchange.close();
+ return;
+ }
+ java.util.regex.Matcher inputArray =
+
java.util.regex.Pattern.compile("\"input\":\\[(.*?)\\]").matcher(LAST_BODY.get());
+ long inputs = inputArray.find() ?
inputArray.group(1).split("\",\"").length : 0;
+ long count = vectorsToReturn >= 0 ? vectorsToReturn : inputs;
+ StringBuilder data = new StringBuilder("{\"data\":[");
+ for (int i = 0; i < count; i++) {
+ data.append(i > 0 ? "," : "").append("{\"index\":").append(i)
+ .append(",\"embedding\":[1.5,-2.0]}");
+ }
+ byte[] response =
data.append("]}").toString().getBytes(StandardCharsets.UTF_8);
+ exchange.sendResponseHeaders(200, response.length);
+ try (OutputStream out = exchange.getResponseBody()) {
+ out.write(response);
+ }
+ });
+ server.start();
+ }
+
+ @AfterAll
+ public static void stopServer() {
+ server.stop(0);
+ }
+
+ @BeforeEach
+ public void reset() {
+ REQUEST_COUNT.set(0);
+ REMAINING_FAILURES.set(0);
+ failureStatus = 500;
+ abortConnection = false;
+ vectorsToReturn = -1;
+ }
+
+ private OpenAICompatibleEmbeddingProvider provider(String apiKeyEnv) {
+ TypedProperties props = new TypedProperties();
+ props.setProperty(EmbeddingTransformerConfig.ENDPOINT_URL.key(),
+ "http://localhost:"; + server.getAddress().getPort() +
"/v1/embeddings");
+ props.setProperty(EmbeddingTransformerConfig.MODEL.key(), "stub-model");
+ if (apiKeyEnv != null) {
+ props.setProperty(EmbeddingTransformerConfig.API_KEY_ENV.key(),
apiKeyEnv);
+ }
+ OpenAICompatibleEmbeddingProvider provider = new
OpenAICompatibleEmbeddingProvider();
+ provider.init(props);
+ return provider;
+ }
+
+ @Test
+ public void testRequestShapeAndVectorParsing() {
+ List<float[]> vectors = provider(null).embed(Arrays.asList("first text",
"second"));
+ assertEquals(2, vectors.size());
+ assertArrayEquals(new float[] {1.5f, -2.0f}, vectors.get(0));
+ assertTrue(LAST_BODY.get().contains("\"model\":\"stub-model\""));
+ assertTrue(LAST_BODY.get().contains("\"input\":[\"first
text\",\"second\"]"));
+ }
+
+ @Test
+ public void testBearerHeaderFromEnvironmentAndMissingEnvFails() {
+ // PATH is always present: its value must be forwarded as the bearer token
+ provider("PATH").embed(Arrays.asList("authed"));
+ assertEquals("Bearer " + System.getenv("PATH"), LAST_AUTH.get());
+
+ assertThrows(HoodieException.class,
+ () ->
provider("HOODIE_TEST_NO_SUCH_ENV_VAR").embed(Arrays.asList("x")));
+ }
+
+ @Test
+ public void testRetriesTransientFailuresHonoringRetryAfter() {
+ // two 503s (Retry-After: 1) then success
+ REMAINING_FAILURES.set(2);
+ failureStatus = 503;
+ long startMs = System.currentTimeMillis();
+ assertEquals(1, provider(null).embed(Arrays.asList("flaky")).size());
+ assertEquals(3, REQUEST_COUNT.get());
+ assertTrue(System.currentTimeMillis() - startMs >= 2000); // two
Retry-After sleeps
+
+ // connection aborted mid-request retries the same way
+ REMAINING_FAILURES.set(1);
+ abortConnection = true;
+ assertEquals(1, provider(null).embed(Arrays.asList("dropped")).size());
+ }
+
+ @Test
+ public void testFailFastOn4xxAndExhaustionOn5xx() {
+ // non-retriable client error fails on the first attempt
+ REMAINING_FAILURES.set(Integer.MAX_VALUE);
+ failureStatus = 404;
+ assertThrows(HoodieException.class, () ->
provider(null).embed(Arrays.asList("gone")));
+ assertEquals(1, REQUEST_COUNT.get());
+
+ // persistent 5xx exhausts all attempts then fails
+ REQUEST_COUNT.set(0);
+ failureStatus = 502;
+ assertThrows(HoodieException.class, () ->
provider(null).embed(Arrays.asList("down")));
+ assertEquals(5, REQUEST_COUNT.get());
+ }
+
+ @Test
+ public void testResponseCountMismatchFails() {
+ vectorsToReturn = 1;
+ assertThrows(HoodieException.class,
+ () -> provider(null).embed(Arrays.asList("one", "two")));
+ }
+}
diff --git a/packaging/hudi-utilities-bundle/pom.xml
b/packaging/hudi-utilities-bundle/pom.xml
index bd2a8b3d91ed..9312ab59d8d7 100644
--- a/packaging/hudi-utilities-bundle/pom.xml
+++ b/packaging/hudi-utilities-bundle/pom.xml
@@ -96,6 +96,7 @@
<include>org.apache.hudi:hudi-client-common</include>
<include>org.apache.hudi:hudi-spark-client</include>
<include>org.apache.hudi:hudi-utilities_${scala.binary.version}</include>
+ <include>org.apache.tika:tika-core</include>
<include>org.apache.hudi:hudi-spark-common_${scala.binary.version}</include>
<include>org.apache.hudi:hudi-spark_${scala.binary.version}</include>
<include>org.apache.hudi:${hudi.spark.module}_${scala.binary.version}</include>
diff --git a/packaging/hudi-utilities-slim-bundle/pom.xml
b/packaging/hudi-utilities-slim-bundle/pom.xml
index bdfa9bc1fceb..623c90294bd9 100644
--- a/packaging/hudi-utilities-slim-bundle/pom.xml
+++ b/packaging/hudi-utilities-slim-bundle/pom.xml
@@ -92,6 +92,7 @@
<artifactSet>
<includes combine.children="append">
<include>org.apache.hudi:hudi-utilities_${scala.binary.version}</include>
+ <include>org.apache.tika:tika-core</include>
<include>org.antlr:stringtemplate</include>
<!-- SPARK-43489 Spark 3.5+ has marked protobuf as provided
-->
diff --git a/pom.xml b/pom.xml
index 3dfa0e05016d..84101f5a0a23 100644
--- a/pom.xml
+++ b/pom.xml
@@ -106,6 +106,7 @@
<fasterxml.jackson.dataformat.yaml.version>${fasterxml.spark3.version}</fasterxml.jackson.dataformat.yaml.version>
<kafka.version>2.0.0</kafka.version>
<pulsar.version>3.0.2</pulsar.version>
+ <tika.version>3.3.2</tika.version>
<kafka.connect.api.version>2.5.0</kafka.connect.api.version>
<kafka.spark3.version>2.8.2</kafka.spark3.version>
<pulsar.spark.version>${pulsar.spark.scala12.version}</pulsar.spark.version>
