You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by "rahil-c (via GitHub)" <gi...@apache.org> on 2025/10/22 23:50:46 UTC
[PR] Implement SparkColumnarFileReader for Datasource integration with Lance [hudi]
rahil-c opened a new pull request, #14135:
URL: https://github.com/apache/hudi/pull/14135
### Describe the issue this Pull Request addresses
Feature: https://github.com/apache/hudi/issues/14127
Discussion: https://github.com/apache/hudi/discussions/14128
#### Note (this is the prior stacked pr https://github.com/apache/hudi/pull/14132, would recommend viewing the prior pr first before viewing this pr)
Implement the reader trait `SparkColumnarFileReader` which serves as a building block for reading the base files in the spark datasource, interface here: https://github.com/apache/hudi/blob/master/hudi-client/hudi-spark-client/src/main/scala/org/apache/spark/sql/execution/datasources/SparkColumnarFileReader.scala
### Summary and Changelog
* Add an implementation of the `SparkColumnarFileReader` for lance which will be`SparkLanceReaderBase`
* Refactored the `LanceIterator` to be reused amongst both file reader and spark reader classes.
* Plumbing thru various FileFactory readers, and spark adapter for supporting lance
* end to end functional test `TestLanceDataSource.scala` to confirm read queries work
### Impact
None
### Risk Level
Low
### Documentation Update
None
### Contributor's checklist
- [ ] Read through [contributor's guide](https://hudi.apache.org/contribute/how-to-contribute)
- [ ] Enough context is provided in the sections above
- [ ] Adequate tests were added if applicable
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Implement SparkColumnarFileReader for Datasource integration with Lance [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #14135:
URL: https://github.com/apache/hudi/pull/14135#issuecomment-3456733625
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "2ddcfdf0df44bda512d701f68911b9377d102f9b",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=9274",
"triggerID" : "2ddcfdf0df44bda512d701f68911b9377d102f9b",
"triggerType" : "PUSH"
}, {
"hash" : "f03d3769c6e20d37d4a90781099d1c4b5d45ffcb",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=9298",
"triggerID" : "f03d3769c6e20d37d4a90781099d1c4b5d45ffcb",
"triggerType" : "PUSH"
}, {
"hash" : "1ceb019de4ab6047cc446596ea456597bdc34201",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "1ceb019de4ab6047cc446596ea456597bdc34201",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* f03d3769c6e20d37d4a90781099d1c4b5d45ffcb Azure: [FAILURE](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=9298)
* 1ceb019de4ab6047cc446596ea456597bdc34201 UNKNOWN
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Implement SparkColumnarFileReader for Datasource integration with Lance [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #14135:
URL: https://github.com/apache/hudi/pull/14135#issuecomment-3434584442
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "2ddcfdf0df44bda512d701f68911b9377d102f9b",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "2ddcfdf0df44bda512d701f68911b9377d102f9b",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 2ddcfdf0df44bda512d701f68911b9377d102f9b UNKNOWN
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Implement SparkColumnarFileReader for Datasource integration with Lance [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #14135:
URL: https://github.com/apache/hudi/pull/14135#issuecomment-3434588333
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "2ddcfdf0df44bda512d701f68911b9377d102f9b",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=9274",
"triggerID" : "2ddcfdf0df44bda512d701f68911b9377d102f9b",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 2ddcfdf0df44bda512d701f68911b9377d102f9b Azure: [FAILURE](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=9274)
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Implement SparkColumnarFileReader for Datasource integration with Lance [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #14135:
URL: https://github.com/apache/hudi/pull/14135#issuecomment-3456759705
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "2ddcfdf0df44bda512d701f68911b9377d102f9b",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=9274",
"triggerID" : "2ddcfdf0df44bda512d701f68911b9377d102f9b",
"triggerType" : "PUSH"
}, {
"hash" : "f03d3769c6e20d37d4a90781099d1c4b5d45ffcb",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=9298",
"triggerID" : "f03d3769c6e20d37d4a90781099d1c4b5d45ffcb",
"triggerType" : "PUSH"
}, {
"hash" : "1ceb019de4ab6047cc446596ea456597bdc34201",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=9385",
"triggerID" : "1ceb019de4ab6047cc446596ea456597bdc34201",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 1ceb019de4ab6047cc446596ea456597bdc34201 Azure: [FAILURE](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=9385)
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Implement SparkColumnarFileReader for Datasource integration with Lance [hudi]
Posted by "rahil-c (via GitHub)" <gi...@apache.org>.
rahil-c commented on PR #14135:
URL: https://github.com/apache/hudi/pull/14135#issuecomment-3434759844
cc @the-other-tim-brown @vinothchandar @yihua @nsivabalan
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Implement SparkColumnarFileReader for Datasource integration with Lance [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #14135:
URL: https://github.com/apache/hudi/pull/14135#issuecomment-3437309840
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "2ddcfdf0df44bda512d701f68911b9377d102f9b",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=9274",
"triggerID" : "2ddcfdf0df44bda512d701f68911b9377d102f9b",
"triggerType" : "PUSH"
}, {
"hash" : "f03d3769c6e20d37d4a90781099d1c4b5d45ffcb",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "f03d3769c6e20d37d4a90781099d1c4b5d45ffcb",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 2ddcfdf0df44bda512d701f68911b9377d102f9b Azure: [FAILURE](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=9274)
* f03d3769c6e20d37d4a90781099d1c4b5d45ffcb UNKNOWN
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Implement SparkColumnarFileReader for Datasource integration with Lance [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #14135:
URL: https://github.com/apache/hudi/pull/14135#issuecomment-3437334463
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "2ddcfdf0df44bda512d701f68911b9377d102f9b",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=9274",
"triggerID" : "2ddcfdf0df44bda512d701f68911b9377d102f9b",
"triggerType" : "PUSH"
}, {
"hash" : "f03d3769c6e20d37d4a90781099d1c4b5d45ffcb",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=9298",
"triggerID" : "f03d3769c6e20d37d4a90781099d1c4b5d45ffcb",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* f03d3769c6e20d37d4a90781099d1c4b5d45ffcb Azure: [FAILURE](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=9298)
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Implement SparkColumnarFileReader for Datasource integration with Lance [hudi]
Posted by "the-other-tim-brown (via GitHub)" <gi...@apache.org>.
the-other-tim-brown commented on code in PR #14135:
URL: https://github.com/apache/hudi/pull/14135#discussion_r2514736168
##########
hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/io/storage/HoodieSparkLanceReader.java:
##########
@@ -0,0 +1,212 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.io.storage;
+
+import com.lancedb.lance.file.LanceFileReader;
+import org.apache.arrow.memory.BufferAllocator;
+import org.apache.arrow.memory.RootAllocator;
+import org.apache.arrow.vector.ipc.ArrowReader;
+import org.apache.hudi.AvroConversionUtils;
+import org.apache.hudi.SparkAdapterSupport$;
+import org.apache.hudi.avro.HoodieAvroUtils;
+import org.apache.hudi.common.bloom.BloomFilter;
+import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.common.model.HoodieSparkRecord;
+import org.apache.hudi.common.util.collection.ClosableIterator;
+import org.apache.hudi.common.util.collection.CloseableMappingIterator;
+import org.apache.hudi.common.util.collection.Pair;
+import org.apache.hudi.exception.HoodieException;
+import org.apache.hudi.exception.HoodieIOException;
+import org.apache.hudi.storage.HoodieStorage;
+import org.apache.hudi.storage.StoragePath;
+import org.apache.spark.sql.catalyst.InternalRow;
+import org.apache.spark.sql.catalyst.expressions.UnsafeRow;
+import org.apache.spark.sql.types.StructField;
+import org.apache.spark.sql.types.StructType;
+import org.apache.spark.sql.util.LanceArrowUtils;
+
+import org.apache.avro.Schema;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+
+import static org.apache.hudi.common.util.TypeUtils.unsafeCast;
+
+/**
+ * {@link HoodieSparkFileReader} implementation for Lance file format.
+ */
+public class HoodieSparkLanceReader implements HoodieSparkFileReader {
+ // number of rows to read
+ private static final int DEFAULT_BATCH_SIZE = 512;
+ private final StoragePath path;
+ private final HoodieStorage storage;
+
+ public HoodieSparkLanceReader(HoodieStorage storage, StoragePath path) {
+ this.path = path;
+ this.storage = storage;
+ }
+
+ @Override
+ public String[] readMinMaxRecordKeys() {
+ throw new UnsupportedOperationException("Min/max record key tracking is not yet supported for Lance file format");
+ }
+
+ @Override
+ public BloomFilter readBloomFilter() {
+ throw new UnsupportedOperationException("Bloom filter is not yet supported for Lance file format");
+ }
+
+ @Override
+ public Set<Pair<String, Long>> filterRowKeys(Set<String> candidateRowKeys) {
+ Set<Pair<String, Long>> result = new HashSet<>();
+ long position = 0;
+
+ try (ClosableIterator<String> keyIterator = getRecordKeyIterator()) {
+ while (keyIterator.hasNext()) {
+ String recordKey = keyIterator.next();
+ // If filter is empty/null, then all keys will be added.
+ // if filter has specific keys, then ensure only those are added
+ if (candidateRowKeys == null || candidateRowKeys.isEmpty()
+ || candidateRowKeys.contains(recordKey)) {
+ result.add(Pair.of(recordKey, position));
+ }
+ position++;
+ }
+ } catch (IOException e) {
+ throw new HoodieIOException("Failed to filter row keys from Lance file: " + path, e);
+ }
+
+ return result;
+ }
+
+ @Override
+ public ClosableIterator<HoodieRecord<InternalRow>> getRecordIterator(Schema readerSchema, Schema requestedSchema) throws IOException {
+ return getRecordIterator(requestedSchema);
+ }
+
+ @Override
+ public ClosableIterator<HoodieRecord<InternalRow>> getRecordIterator(Schema schema) throws IOException {
+ ClosableIterator<UnsafeRow> iterator = getUnsafeRowIterator();
+ return new CloseableMappingIterator<>(iterator, data -> unsafeCast(new HoodieSparkRecord(data)));
+ }
+
+ @Override
+ public ClosableIterator<String> getRecordKeyIterator() throws IOException {
+ //TODO to revisit adding support for when metadata fields are not persisted.
Review Comment:
It looks like the other implementations do not do this, @yihua does this need to be supported?
##########
hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/io/storage/HoodieSparkLanceReader.java:
##########
@@ -0,0 +1,212 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.io.storage;
+
+import com.lancedb.lance.file.LanceFileReader;
+import org.apache.arrow.memory.BufferAllocator;
+import org.apache.arrow.memory.RootAllocator;
+import org.apache.arrow.vector.ipc.ArrowReader;
+import org.apache.hudi.AvroConversionUtils;
+import org.apache.hudi.SparkAdapterSupport$;
+import org.apache.hudi.avro.HoodieAvroUtils;
+import org.apache.hudi.common.bloom.BloomFilter;
+import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.common.model.HoodieSparkRecord;
+import org.apache.hudi.common.util.collection.ClosableIterator;
+import org.apache.hudi.common.util.collection.CloseableMappingIterator;
+import org.apache.hudi.common.util.collection.Pair;
+import org.apache.hudi.exception.HoodieException;
+import org.apache.hudi.exception.HoodieIOException;
+import org.apache.hudi.storage.HoodieStorage;
+import org.apache.hudi.storage.StoragePath;
+import org.apache.spark.sql.catalyst.InternalRow;
+import org.apache.spark.sql.catalyst.expressions.UnsafeRow;
+import org.apache.spark.sql.types.StructField;
+import org.apache.spark.sql.types.StructType;
+import org.apache.spark.sql.util.LanceArrowUtils;
+
+import org.apache.avro.Schema;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+
+import static org.apache.hudi.common.util.TypeUtils.unsafeCast;
+
+/**
+ * {@link HoodieSparkFileReader} implementation for Lance file format.
+ */
+public class HoodieSparkLanceReader implements HoodieSparkFileReader {
+ // number of rows to read
+ private static final int DEFAULT_BATCH_SIZE = 512;
+ private final StoragePath path;
+ private final HoodieStorage storage;
+
+ public HoodieSparkLanceReader(HoodieStorage storage, StoragePath path) {
+ this.path = path;
+ this.storage = storage;
+ }
+
+ @Override
+ public String[] readMinMaxRecordKeys() {
+ throw new UnsupportedOperationException("Min/max record key tracking is not yet supported for Lance file format");
+ }
+
+ @Override
+ public BloomFilter readBloomFilter() {
+ throw new UnsupportedOperationException("Bloom filter is not yet supported for Lance file format");
+ }
+
+ @Override
+ public Set<Pair<String, Long>> filterRowKeys(Set<String> candidateRowKeys) {
+ Set<Pair<String, Long>> result = new HashSet<>();
+ long position = 0;
+
+ try (ClosableIterator<String> keyIterator = getRecordKeyIterator()) {
+ while (keyIterator.hasNext()) {
+ String recordKey = keyIterator.next();
+ // If filter is empty/null, then all keys will be added.
+ // if filter has specific keys, then ensure only those are added
+ if (candidateRowKeys == null || candidateRowKeys.isEmpty()
+ || candidateRowKeys.contains(recordKey)) {
+ result.add(Pair.of(recordKey, position));
+ }
+ position++;
+ }
+ } catch (IOException e) {
+ throw new HoodieIOException("Failed to filter row keys from Lance file: " + path, e);
+ }
+
+ return result;
+ }
+
+ @Override
+ public ClosableIterator<HoodieRecord<InternalRow>> getRecordIterator(Schema readerSchema, Schema requestedSchema) throws IOException {
+ return getRecordIterator(requestedSchema);
+ }
+
+ @Override
+ public ClosableIterator<HoodieRecord<InternalRow>> getRecordIterator(Schema schema) throws IOException {
+ ClosableIterator<UnsafeRow> iterator = getUnsafeRowIterator();
+ return new CloseableMappingIterator<>(iterator, data -> unsafeCast(new HoodieSparkRecord(data)));
+ }
+
+ @Override
+ public ClosableIterator<String> getRecordKeyIterator() throws IOException {
+ //TODO to revisit adding support for when metadata fields are not persisted.
+
+ // Get schema with only the record key field for efficient column pruning
+ Schema recordKeySchema = HoodieAvroUtils.getRecordKeySchema();
+ ClosableIterator<UnsafeRow> iterator = getUnsafeRowIterator(recordKeySchema);
+
+ // Map each UnsafeRow to extract the record key string directly from index 0
+ // The record key is at index 0 because we're using lance column projection which has only the record key field
+ return new CloseableMappingIterator<>(iterator, data -> data.getUTF8String(0).toString());
+ }
+
+ /**
+ * Get an iterator over UnsafeRows from the Lance file.
+ *
+ * @return ClosableIterator over UnsafeRows
+ * @throws IOException if reading fails
+ */
+ public ClosableIterator<UnsafeRow> getUnsafeRowIterator() {
+ BufferAllocator allocator = new RootAllocator(Long.MAX_VALUE);
+
+ try {
+ LanceFileReader lanceReader = LanceFileReader.open(path.toString(), allocator);
+
+ // Get schema from Lance file and convert to Spark StructType
+ org.apache.arrow.vector.types.pojo.Schema arrowSchema = lanceReader.schema();
+ StructType sparkSchema = LanceArrowUtils.fromArrowSchema(arrowSchema);
+
+ ArrowReader arrowReader = lanceReader.readAll(null, null, DEFAULT_BATCH_SIZE);
+
+ return new HoodieLanceRecordIterator(allocator, lanceReader, arrowReader, sparkSchema, path.toString());
+ } catch (Exception e) {
+ allocator.close();
+ throw new HoodieException("Failed to create Lance reader for: " + path, e);
+ }
+ }
+
+ /**
+ * Get an iterator over UnsafeRows from the Lance file with column projection.
+ * This allows reading only specific columns for better performance.
+ *
+ * @param requestedSchema Avro schema specifying which columns to read
+ * @return ClosableIterator over UnsafeRows
+ * @throws IOException if reading fails
+ */
+ public ClosableIterator<UnsafeRow> getUnsafeRowIterator(Schema requestedSchema) {
+ // Convert Avro schema to Spark StructType
+ StructType requestedSparkSchema = AvroConversionUtils.convertAvroSchemaToStructType(requestedSchema);
+
+ BufferAllocator allocator = new RootAllocator(Long.MAX_VALUE);
+
+ try {
+ LanceFileReader lanceReader = LanceFileReader.open(path.toString(), allocator);
+
+ // Build list of column names to project (read only requested columns)
Review Comment:
Does this work for struct types and only reading back a subset of nested fields?
##########
hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/io/storage/HoodieSparkLanceReader.java:
##########
@@ -0,0 +1,212 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.io.storage;
+
+import com.lancedb.lance.file.LanceFileReader;
+import org.apache.arrow.memory.BufferAllocator;
+import org.apache.arrow.memory.RootAllocator;
+import org.apache.arrow.vector.ipc.ArrowReader;
+import org.apache.hudi.AvroConversionUtils;
+import org.apache.hudi.SparkAdapterSupport$;
+import org.apache.hudi.avro.HoodieAvroUtils;
+import org.apache.hudi.common.bloom.BloomFilter;
+import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.common.model.HoodieSparkRecord;
+import org.apache.hudi.common.util.collection.ClosableIterator;
+import org.apache.hudi.common.util.collection.CloseableMappingIterator;
+import org.apache.hudi.common.util.collection.Pair;
+import org.apache.hudi.exception.HoodieException;
+import org.apache.hudi.exception.HoodieIOException;
+import org.apache.hudi.storage.HoodieStorage;
+import org.apache.hudi.storage.StoragePath;
+import org.apache.spark.sql.catalyst.InternalRow;
+import org.apache.spark.sql.catalyst.expressions.UnsafeRow;
+import org.apache.spark.sql.types.StructField;
+import org.apache.spark.sql.types.StructType;
+import org.apache.spark.sql.util.LanceArrowUtils;
+
+import org.apache.avro.Schema;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+
+import static org.apache.hudi.common.util.TypeUtils.unsafeCast;
+
+/**
+ * {@link HoodieSparkFileReader} implementation for Lance file format.
+ */
+public class HoodieSparkLanceReader implements HoodieSparkFileReader {
+ // number of rows to read
+ private static final int DEFAULT_BATCH_SIZE = 512;
+ private final StoragePath path;
+ private final HoodieStorage storage;
+
+ public HoodieSparkLanceReader(HoodieStorage storage, StoragePath path) {
+ this.path = path;
+ this.storage = storage;
+ }
+
+ @Override
+ public String[] readMinMaxRecordKeys() {
+ throw new UnsupportedOperationException("Min/max record key tracking is not yet supported for Lance file format");
Review Comment:
Let's make sure there is a subtask for these unsupported operations if there isn't already.
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Implement SparkColumnarFileReader for Datasource integration with Lance [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #14135:
URL: https://github.com/apache/hudi/pull/14135#issuecomment-3434586422
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "2ddcfdf0df44bda512d701f68911b9377d102f9b",
"status" : "PENDING",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=9274",
"triggerID" : "2ddcfdf0df44bda512d701f68911b9377d102f9b",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 2ddcfdf0df44bda512d701f68911b9377d102f9b Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=9274)
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Implement SparkColumnarFileReader for Datasource integration with Lance [hudi]
Posted by "rahil-c (via GitHub)" <gi...@apache.org>.
rahil-c commented on PR #14135:
URL: https://github.com/apache/hudi/pull/14135#issuecomment-3462255595
Currently the test lance data source is failing in the CI. Locally however it passes
<img width="2473" height="481" alt="Screenshot 2025-10-29 at 8 28 31 AM" src="https://github.com/user-attachments/assets/b9263a8d-8f06-4f5a-82e6-23402c2abf8c" />
The reason for the ci failure is currently I only have lance support for spark 3.5 so spark versions less than that should not run. Will need to make this change
```
[ERROR] testBasicWriteAndRead Time elapsed: 1.869 s <<< ERROR!
2025-10-28T14:34:40.3914762Z java.lang.UnsupportedOperationException: Lance format is not supported in Spark 3.3
2025-10-28T14:34:40.3915807Z at org.apache.spark.sql.adapter.Spark3_3Adapter.createLanceFileReader(Spark3_3Adapter.scala:158)
2025-10-28T14:34:40.3917446Z at org.apache.spark.sql.execution.datasources.parquet.HoodieFileGroupReaderBasedFileFormat.buildBaseFileReader(HoodieFileGroupReaderBasedFileFormat.scala:298)
2025-10-28T14:34:40.3919142Z at org.apache.spark.sql.execution.datasources.parquet.HoodieFileGroupReaderBasedFileFormat.buildReaderWithPartitionValues(HoodieFileGroupReaderBasedFileFormat.scala:209)
2025-10-28T14:34:40.3920159Z at org.apache.spark.sql.execution.FileSourceScanExec.inputRDD$lzycompute(DataSourceScanExec.scala:465)
2025-10-28T14:34:40.3920808Z at org.apache.spark.sql.execution.FileSourceScanExec.inputRDD(DataSourceScanExec.scala:454)
2025-10-28T14:34:40.3921415Z at org.apache.spark.sql.execution.FileSourceScanExec.doExecute(DataSourceScanExec.scala:534)
2025-10-28T14:34:40.3921986Z at org.apache.spark.sql.execution.SparkPlan.$anonfun$execute$1(SparkPlan.scala:194)
2025-10-28T14:34:40.3922518Z at org.apache.spark.sql.execution.SparkPlan.$anonfun$executeQuery$1(SparkPlan.scala:232)
2025-10-28T14:34:40.3923042Z at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:151)
2025-10-28T14:34:40.3924027Z at org.apache.spark.sql.execution.SparkPlan.executeQuery(SparkPlan.scala:229)
2025-10-28T14:34:40.3924512Z at org.apache.spark.sql.execution.SparkPlan.execute(SparkPlan.scala:190)
2025-10-28T14:34:40.3925133Z at org.apache.spark.sql.execution.exchange.ShuffleExchangeExec.inputRDD$lzycompute(ShuffleExchangeExec.scala:135)
2025-10-28T14:34:40.3925869Z at org.apache.spark.sql.execution.exchange.ShuffleExchangeExec.inputRDD(ShuffleExchangeExec.scala:135)
2025-10-28T14:34:40.3926679Z at org.apache.spark.sql.execution.exchange.ShuffleExchangeExec.mapOutputStatisticsFuture$lzycompute(ShuffleExchangeExec.scala:140)
2025-10-28T14:34:40.3927572Z at org.apache.spark.sql.execution.exchange.ShuffleExchangeExec.mapOutputStatisticsFuture(ShuffleExchangeExec.scala:139)
2025-10-28T14:34:40.3928387Z at org.apache.spark.sql.execution.exchange.ShuffleExchangeLike.$anonfun$submitShuffleJob$1(ShuffleExchangeExec.scala:68)
2025-10-28T14:34:40.3929054Z at org.apache.spark.sql.execution.SparkPlan.$anonfun$executeQuery$1(SparkPlan.scala:232)
2025-10-28T14:34:40.3929686Z at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:151)
2025-10-28T14:34:40.3930176Z at org.apache.spark.sql.execution.SparkPlan.executeQuery(SparkPlan.scala:229)
2025-10-28T14:34:40.3930797Z at org.apache.spark.sql.execution.exchange.ShuffleExchangeLike.submitShuffleJob(ShuffleExchangeExec.scala:68)
2025-10-28T14:34:40.3931565Z at org.apache.spark.sql.execution.exchange.ShuffleExchangeLike.submitShuffleJob$(ShuffleExchangeExec.scala:67)
2025-10-28T14:34:40.3932329Z at org.apache.spark.sql.execution.exchange.ShuffleExchangeExec.submitShuffleJob(ShuffleExchangeExec.scala:115)
2025-10-28T14:34:40.3933105Z at org.apache.spark.sql.execution.adaptive.ShuffleQueryStageExec.shuffleFuture$lzycompute(QueryStageExec.scala:174)
2025-10-28T14:34:40.3934050Z at org.apache.spark.sql.execution.adaptive.ShuffleQueryStageExec.shuffleFuture(QueryStageExec.scala:174)
2025-10-28T14:34:40.3934767Z at org.apache.spark.sql.execution.adaptive.ShuffleQueryStageExec.doMaterialize(QueryStageExec.scala:176)
2025-10-28T14:34:40.3935431Z at org.apache.spark.sql.execution.adaptive.QueryStageExec.materialize(QueryStageExec.scala:82)
2025-10-28T14:34:40.3936343Z at org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanExec.$anonfun$getFinalPhysicalPlan$5(AdaptiveSparkPlanExec.scala:263)
2025-10-28T14:34:40.3937246Z at org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanExec.$anonfun$getFinalPhysicalPlan$5$adapted(AdaptiveSparkPlanExec.scala:261)
2025-10-28T14:34:40.3937884Z at scala.collection.Iterator.foreach(Iterator.scala:943)
2025-10-28T14:34:40.3938238Z at scala.collection.Iterator.foreach$(Iterator.scala:943)
2025-10-28T14:34:40.3938677Z at scala.collection.AbstractIterator.foreach(Iterator.scala:1431)
2025-10-28T14:34:40.3939085Z at scala.collection.IterableLike.foreach(IterableLike.scala:74)
2025-10-28T14:34:40.3939479Z at scala.collection.IterableLike.foreach$(IterableLike.scala:73)
2025-10-28T14:34:40.3939869Z at scala.collection.AbstractIterable.foreach(Iterable.scala:56)
2025-10-28T14:34:40.3940499Z at org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanExec.$anonfun$getFinalPhysicalPlan$1(AdaptiveSparkPlanExec.scala:261)
2025-10-28T14:34:40.3941164Z at org.apache.spark.sql.SparkSession.withActive(SparkSession.scala:779)
2025-10-28T14:34:40.3941814Z at org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanExec.getFinalPhysicalPlan(AdaptiveSparkPlanExec.scala:233)
2025-10-28T14:34:40.3942653Z at org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanExec.withFinalPlanUpdate(AdaptiveSparkPlanExec.scala:375)
2025-10-28T14:34:40.3943628Z at org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanExec.executeCollect(AdaptiveSparkPlanExec.scala:348)
2025-10-28T14:34:40.3944248Z at org.apache.spark.sql.Dataset.collectFromPlan(Dataset.scala:3908)
2025-10-28T14:34:40.3944665Z at org.apache.spark.sql.Dataset.$anonfun$collect$1(Dataset.scala:3160)
2025-10-28T14:34:40.3945171Z at org.apache.spark.sql.Dataset.$anonfun$withAction$2(Dataset.scala:3898)
2025-10-28T14:34:40.3945687Z at org.apache.spark.sql.execution.QueryExecution$.withInternalError(QueryExecution.scala:510)
2025-10-28T14:34:40.3946214Z at org.apache.spark.sql.Dataset.$anonfun$withAction$1(Dataset.scala:3896)
2025-10-28T14:34:40.3946747Z at org.apache.spark.sql.execution.SQLExecution$.$anonfun$withNewExecutionId$6(SQLExecution.scala:109)
2025-10-28T14:34:40.3947367Z at org.apache.spark.sql.execution.SQLExecution$.withSQLConfPropagated(SQLExecution.scala:169)
2025-10-28T14:34:40.3947975Z at org.apache.spark.sql.execution.SQLExecution$.$anonfun$withNewExecutionId$1(SQLExecution.scala:95)
2025-10-28T14:34:40.3948508Z at org.apache.spark.sql.SparkSession.withActive(SparkSession.scala:779)
2025-10-28T14:34:40.3949004Z at org.apache.spark.sql.execution.SQLExecution$.withNewExecutionId(SQLExecution.scala:64)
2025-10-28T14:34:40.3949478Z at org.apache.spark.sql.Dataset.withAction(Dataset.scala:3896)
2025-10-28T14:34:40.3949852Z at org.apache.spark.sql.Dataset.collect(Dataset.scala:3160)
2025-10-28T14:34:40.3950454Z at org.apache.hudi.functional.TestLanceDataSource.testBasicWriteAndRead(TestLanceDataSource.scala:80)
2025-10-28T14:34:40.3951010Z at java.base/java.lang.reflect.Method.invoke(Method.java:569)
2025-10-28T14:34:40.3951378Z at java.base/java.util.ArrayList.forEach(ArrayList.java:1511)
2025-10-28T14:34:40.3951743Z at java.base/java.util.ArrayList.forEach(ArrayList.java:1511)
```
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Implement SparkColumnarFileReader for Datasource integration with Lance [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #14135:
URL: https://github.com/apache/hudi/pull/14135#issuecomment-3437318007
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "2ddcfdf0df44bda512d701f68911b9377d102f9b",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=9274",
"triggerID" : "2ddcfdf0df44bda512d701f68911b9377d102f9b",
"triggerType" : "PUSH"
}, {
"hash" : "f03d3769c6e20d37d4a90781099d1c4b5d45ffcb",
"status" : "PENDING",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=9298",
"triggerID" : "f03d3769c6e20d37d4a90781099d1c4b5d45ffcb",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 2ddcfdf0df44bda512d701f68911b9377d102f9b Azure: [FAILURE](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=9274)
* f03d3769c6e20d37d4a90781099d1c4b5d45ffcb Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=9298)
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Implement SparkColumnarFileReader for Datasource integration with Lance [hudi]
Posted by "rahil-c (via GitHub)" <gi...@apache.org>.
rahil-c closed pull request #14135: Implement SparkColumnarFileReader for Datasource integration with Lance
URL: https://github.com/apache/hudi/pull/14135
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Implement SparkColumnarFileReader for Datasource integration with Lance [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #14135:
URL: https://github.com/apache/hudi/pull/14135#issuecomment-3456743569
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "2ddcfdf0df44bda512d701f68911b9377d102f9b",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=9274",
"triggerID" : "2ddcfdf0df44bda512d701f68911b9377d102f9b",
"triggerType" : "PUSH"
}, {
"hash" : "f03d3769c6e20d37d4a90781099d1c4b5d45ffcb",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=9298",
"triggerID" : "f03d3769c6e20d37d4a90781099d1c4b5d45ffcb",
"triggerType" : "PUSH"
}, {
"hash" : "1ceb019de4ab6047cc446596ea456597bdc34201",
"status" : "PENDING",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=9385",
"triggerID" : "1ceb019de4ab6047cc446596ea456597bdc34201",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* f03d3769c6e20d37d4a90781099d1c4b5d45ffcb Azure: [FAILURE](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=9298)
* 1ceb019de4ab6047cc446596ea456597bdc34201 Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=9385)
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org