vinothchandar commented on code in PR #19278: URL: https://github.com/apache/hudi/pull/19278#discussion_r3651691653
########## hudi-utilities/src/main/java/org/apache/hudi/utilities/config/UnstructuredFileSourceConfig.java: ########## @@ -0,0 +1,112 @@ +/* + * 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; + +/** + * 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> PARSER_CLASS = ConfigProperty Review Comment: turn this into an enum. and change this to `DOCUMENT_PARSER` . we can internally map a value `tika` to the current class. It can help us add few blessed classes like this for now. ########## hudi-utilities/src/main/java/org/apache/hudi/utilities/config/EmbeddingTransformerConfig.java: ########## @@ -0,0 +1,121 @@ +/* + * 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", Review Comment: verify these groups are correct ########## hudi-utilities/src/main/java/org/apache/hudi/utilities/config/UnstructuredFileSourceConfig.java: ########## @@ -0,0 +1,112 @@ +/* + * 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; + +/** + * 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> PARSER_CLASS = ConfigProperty + .key(PREFIX + "parser.class") + .defaultValue("org.apache.hudi.utilities.sources.helpers.unstructured.TikaDocumentParser") + .markAdvanced() + .sinceVersion("1.2.0") + .withDocumentation("Implementation of DocumentParser used to extract text and metadata " + + "from ingested files. The default uses Apache Tika with automatic format detection."); + + 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."); + + 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 Review Comment: this needs to be as much as num_executors * num_cores? ########## hudi-utilities/src/main/java/org/apache/hudi/utilities/config/UnstructuredFileSourceConfig.java: ########## @@ -0,0 +1,112 @@ +/* + * 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; + +/** + * 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> PARSER_CLASS = ConfigProperty + .key(PREFIX + "parser.class") + .defaultValue("org.apache.hudi.utilities.sources.helpers.unstructured.TikaDocumentParser") + .markAdvanced() + .sinceVersion("1.2.0") + .withDocumentation("Implementation of DocumentParser used to extract text and metadata " + + "from ingested files. The default uses Apache Tika with automatic format detection."); + + 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 Review Comment: would it skip `.parquet`, `.orc`, `.avro` etc? whats the behavior with a mix of structured and unstructured cols. Should be ignore these columnar or data file formats by default ########## 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 { Review Comment: does this have enough UTs. cover the basic file types? ########## pom.xml: ########## @@ -107,6 +107,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>2.9.4</tika.version> Review Comment: can't we use a later version https://tika.apache.org/3.3.2/index.html ########## hudi-utilities/src/main/java/org/apache/hudi/utilities/transform/embedding/OpenAICompatibleEmbeddingProvider.java: ########## @@ -0,0 +1,178 @@ +/* + * 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); + } + } + } + + private 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() Review Comment: lets ensure connections are pooled and not created each time? that will be very slow/inefficient. and also we should ideally parallelize here? i.e we are not blocking on responses? I wonder if we fire http requests async and collect responses asthey come in, having some thread pool for requests.. ########## hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/helpers/unstructured/ParseResult.java: ########## @@ -0,0 +1,92 @@ +/* + * 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 java.io.Serializable; +import java.util.Collections; +import java.util.Map; + +/** + * Outcome of parsing one file with a {@link DocumentParser}. + */ +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); + } + + public ParseStatus getStatus() { Review Comment: use lombok? ########## hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/helpers/unstructured/TextChunker.java: ########## @@ -0,0 +1,80 @@ +/* + * 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 fixed-size, overlapping character chunks for + * retrieval-oriented consumers. + */ +public class TextChunker implements Serializable { + + private static final long serialVersionUID = 1L; + + private final int chunkSizeChars; + private final int overlapChars; + + public TextChunker(int chunkSizeChars, int overlapChars) { Review Comment: can we do sth smart like chunking at sentence or paragraph boundary? what the state of the art or norm around text chunking. its good to follow best practices ########## hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/UnstructuredFileDFSSource.java: ########## @@ -0,0 +1,170 @@ +/* + * 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.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, Review Comment: can we make this work with S3IncrSource and GCSIncrSource as well i.e the modification-time based incremental has its scaling limits. we should make this part configurable? or have diffrent sources while sharing code without much duplication ########## hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/helpers/unstructured/UnstructuredFileRecordBuilder.java: ########## @@ -0,0 +1,150 @@ +/* + * 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.common.util.ConfigUtils.getStringWithAltKeys; +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.PARSER_CLASS; +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 = getStringWithAltKeys(props, PARSER_CLASS, true); + 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 { + byte[] bytes = new byte[size]; Review Comment: can we use off heap bytebuffers instead of stressing JVM with large allocations? the byte buffers need to be released cleanly (closeable iteraotr?) and also we some doc/comment on increased overhead memory required now ########## hudi-utilities/src/main/java/org/apache/hudi/utilities/transform/embedding/OpenAICompatibleEmbeddingProvider.java: ########## @@ -0,0 +1,178 @@ +/* + * 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 { Review Comment: UT this class thoroughly -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
