You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by "hudi-agent (via GitHub)" <gi...@apache.org> on 2026/04/27 03:50:27 UTC
Re: [PR] Add lance schema evolution support for add column [hudi]
hudi-agent commented on code in PR #14160:
URL: https://github.com/apache/hudi/pull/14160#discussion_r3144784452
##########
hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/io/storage/HoodieLanceRecordIterator.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.io.storage;
+
+import com.lancedb.lance.file.LanceFileReader;
+import org.apache.arrow.memory.BufferAllocator;
+import org.apache.arrow.vector.VectorSchemaRoot;
+import org.apache.arrow.vector.ipc.ArrowReader;
+import org.apache.hudi.common.util.collection.ClosableIterator;
+import org.apache.hudi.exception.HoodieException;
+import org.apache.hudi.exception.HoodieIOException;
+import org.apache.spark.sql.catalyst.InternalRow;
+import org.apache.spark.sql.catalyst.expressions.UnsafeProjection;
+import org.apache.spark.sql.catalyst.expressions.UnsafeRow;
+import org.apache.spark.sql.types.StructType;
+import org.apache.spark.sql.vectorized.ColumnVector;
+import org.apache.spark.sql.vectorized.ColumnarBatch;
+import org.apache.spark.sql.vectorized.LanceArrowColumnVector;
+
+import java.io.IOException;
+import java.util.Iterator;
+
+/**
+ * Shared iterator implementation for reading Lance files and converting Arrow batches to Spark rows.
+ * This iterator is used by both Hudi's internal Lance reader and Spark datasource integration.
+ *
+ * <p>The iterator manages the lifecycle of:
+ * <ul>
+ * <li>BufferAllocator - Arrow memory management</li>
+ * <li>LanceFileReader - Lance file handle</li>
+ * <li>ArrowReader - Arrow batch reader</li>
+ * <li>ColumnarBatch - Current batch being iterated</li>
+ * </ul>
+ *
+ * <p>Records are converted to {@link UnsafeRow} using {@link UnsafeProjection} for efficient
+ * serialization and memory management.
+ */
+public class HoodieLanceRecordIterator implements ClosableIterator<UnsafeRow> {
+ private final BufferAllocator allocator;
+ private final LanceFileReader lanceReader;
+ private final ArrowReader arrowReader;
+ private final StructType schema;
+ private final UnsafeProjection projection;
+ private final String path;
+
+ private ColumnarBatch currentBatch;
+ private Iterator<InternalRow> rowIterator;
+ private boolean closed = false;
+
+ /**
+ * Creates a new Lance record iterator.
+ *
+ * @param allocator Arrow buffer allocator for memory management
+ * @param lanceReader Lance file reader
+ * @param arrowReader Arrow reader for batch reading
+ * @param schema Spark schema for the records
+ * @param path File path (for error messages)
+ */
+ public HoodieLanceRecordIterator(BufferAllocator allocator,
+ LanceFileReader lanceReader,
+ ArrowReader arrowReader,
+ StructType schema,
+ String path) {
+ this.allocator = allocator;
+ this.lanceReader = lanceReader;
+ this.arrowReader = arrowReader;
+ this.schema = schema;
+ this.projection = UnsafeProjection.create(schema);
+ this.path = path;
+ }
+
+ @Override
+ public boolean hasNext() {
+ // If we have records in current batch, return true
+ if (rowIterator != null && rowIterator.hasNext()) {
+ return true;
+ }
+
+ // Close previous batch before loading next
+ if (currentBatch != null) {
Review Comment:
🤖 There is already defensive code elsewhere in the repo (`LanceRecordIterator.buildColumnVectors`) that reorders FieldVectors by name because lance-spark's `VectorSchemaRoot` may return columns in the file's on-disk order rather than the requested projection order. Here we simply stream `root.getFieldVectors()` and rely on positional alignment with the `UnsafeProjection` schema. If the projected `columnNames` aren't a prefix in file order (e.g. record-key-only projection on a wide schema, or any reordering for schema evolution), the resulting UnsafeRows will silently contain misaligned data — `getRecordKeyIterator` could return non-key columns, and `SparkLanceReaderBase` would write wrong types into the result row. Could you mirror the name-based reordering used in `LanceRecordIterator`?
<sub><i>- AI-generated; verify before applying. React 👍/👎 to flag quality.</i></sub>
##########
hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/io/storage/HoodieLanceRecordIterator.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.io.storage;
+
+import com.lancedb.lance.file.LanceFileReader;
+import org.apache.arrow.memory.BufferAllocator;
+import org.apache.arrow.vector.VectorSchemaRoot;
+import org.apache.arrow.vector.ipc.ArrowReader;
+import org.apache.hudi.common.util.collection.ClosableIterator;
+import org.apache.hudi.exception.HoodieException;
+import org.apache.hudi.exception.HoodieIOException;
+import org.apache.spark.sql.catalyst.InternalRow;
+import org.apache.spark.sql.catalyst.expressions.UnsafeProjection;
+import org.apache.spark.sql.catalyst.expressions.UnsafeRow;
+import org.apache.spark.sql.types.StructType;
+import org.apache.spark.sql.vectorized.ColumnVector;
+import org.apache.spark.sql.vectorized.ColumnarBatch;
+import org.apache.spark.sql.vectorized.LanceArrowColumnVector;
+
+import java.io.IOException;
+import java.util.Iterator;
+
+/**
+ * Shared iterator implementation for reading Lance files and converting Arrow batches to Spark rows.
+ * This iterator is used by both Hudi's internal Lance reader and Spark datasource integration.
+ *
+ * <p>The iterator manages the lifecycle of:
+ * <ul>
+ * <li>BufferAllocator - Arrow memory management</li>
+ * <li>LanceFileReader - Lance file handle</li>
+ * <li>ArrowReader - Arrow batch reader</li>
+ * <li>ColumnarBatch - Current batch being iterated</li>
+ * </ul>
+ *
+ * <p>Records are converted to {@link UnsafeRow} using {@link UnsafeProjection} for efficient
+ * serialization and memory management.
+ */
+public class HoodieLanceRecordIterator implements ClosableIterator<UnsafeRow> {
+ private final BufferAllocator allocator;
+ private final LanceFileReader lanceReader;
+ private final ArrowReader arrowReader;
+ private final StructType schema;
+ private final UnsafeProjection projection;
+ private final String path;
+
+ private ColumnarBatch currentBatch;
+ private Iterator<InternalRow> rowIterator;
+ private boolean closed = false;
+
+ /**
+ * Creates a new Lance record iterator.
+ *
+ * @param allocator Arrow buffer allocator for memory management
+ * @param lanceReader Lance file reader
+ * @param arrowReader Arrow reader for batch reading
+ * @param schema Spark schema for the records
+ * @param path File path (for error messages)
+ */
+ public HoodieLanceRecordIterator(BufferAllocator allocator,
+ LanceFileReader lanceReader,
+ ArrowReader arrowReader,
+ StructType schema,
+ String path) {
+ this.allocator = allocator;
+ this.lanceReader = lanceReader;
+ this.arrowReader = arrowReader;
+ this.schema = schema;
+ this.projection = UnsafeProjection.create(schema);
+ this.path = path;
+ }
+
+ @Override
+ public boolean hasNext() {
+ // If we have records in current batch, return true
+ if (rowIterator != null && rowIterator.hasNext()) {
+ return true;
+ }
+
+ // Close previous batch before loading next
+ if (currentBatch != null) {
+ currentBatch.close();
+ currentBatch = null;
+ }
+
+ // Try to load next batch
+ try {
Review Comment:
🤖 If `arrowReader.loadNextBatch()` ever returns true with a zero-row batch, this returns `false` and we'll stop iterating even when more batches follow. Should this be a `while` loop that keeps loading until we either get a non-empty batch or `loadNextBatch()` returns false? Worth confirming Lance's contract here — even if it doesn't currently emit empty batches, a `while` loop would be safer.
<sub><i>- AI-generated; verify before applying. React 👍/👎 to flag quality.</i></sub>
##########
hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/io/storage/HoodieSparkLanceReader.java:
##########
@@ -0,0 +1,213 @@
+/*
+ * 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();
+ //TODO .copy() is needed for correctness, to investigate further in future.
+ return new CloseableMappingIterator<>(iterator, data -> unsafeCast(new HoodieSparkRecord(data.copy())));
+ }
+
+ @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)
+ List<String> columnNames = new ArrayList<>();
+ for (StructField field : requestedSparkSchema.fields()) {
+ columnNames.add(field.name());
+ }
+
+ // Read only the requested columns from Lance file for efficiency
Review Comment:
🤖 `getRecordKeyIterator` projects only the record key, but assumes meta fields are always present in the file. If a Lance table is written with `populate.meta.fields=false` (allowed via the writer config), `HoodieAvroUtils.getRecordKeySchema()` will request a column that doesn't exist in the Lance file and Lance will throw. The TODO acknowledges this but the iterator currently has no defensive handling — could we at least throw a clearer error here directing users to enable meta fields, rather than letting the Lance error bubble up?
<sub><i>- AI-generated; verify before applying. React 👍/👎 to flag quality.</i></sub>
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieSparkSqlWriter.scala:
##########
@@ -510,7 +510,9 @@ class HoodieSparkSqlWriterInternal {
val writeConfig = client.getConfig
if (writeConfig.getRecordMerger.getRecordType == HoodieRecordType.SPARK && tableType == MERGE_ON_READ && writeConfig.getLogDataBlockFormat.orElse(HoodieLogBlockType.AVRO_DATA_BLOCK) != HoodieLogBlockType.PARQUET_DATA_BLOCK) {
- throw new UnsupportedOperationException(s"${writeConfig.getRecordMerger.getClass.getName} only support parquet log.")
+ //TODO to revisit validity of this condition
Review Comment:
🤖 Commenting out this check removes the SPARK-record-with-non-PARQUET-log-block guard for ALL MoR tables, not just Lance. This guard was presumably there because `HoodieSparkRecord.getAvroBytes` previously threw `UnsupportedOperationException` (you've now implemented it). Could we either (a) delete the dead code and add a comment justifying why it's safe across all formats now that getAvroBytes works, or (b) narrow the guard to skip only when base format is LANCE? Leaving commented-out code with a TODO in a write-path validator makes future readers nervous about whether this was intended to ship.
<sub><i>- AI-generated; verify before applying. React 👍/👎 to flag quality.</i></sub>
##########
hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/common/model/HoodieSparkRecord.java:
##########
@@ -327,7 +328,18 @@ public Option<HoodieAvroIndexedRecord> toIndexedRecord(Schema recordSchema, Prop
@Override
public ByteArrayOutputStream getAvroBytes(Schema recordSchema, Properties props) throws IOException {
- throw new UnsupportedOperationException();
+ // Convert Spark InternalRow to Avro GenericRecord
+ if (data == null) {
+ throw new IOException("Cannot convert null data to Avro bytes");
+ }
+ // Get or compute the Spark StructType from Avro schema
+ StructType structType = HoodieInternalRowUtils.getCachedSchema(recordSchema);
+ // Convert InternalRow to GenericRecord using existing converter
+ GenericRecord avroRecord = AvroConversionUtils
+ .createInternalRowToAvroConverter(structType, recordSchema, false)
Review Comment:
🤖 `AvroConversionUtils.createInternalRowToAvroConverter(...)` builds a fresh converter on every call to `getAvroBytes`. In the AVRO_DATA_BLOCK write path this gets called per-record, so the converter construction cost will dominate. Could we cache the converter per (structType, recordSchema) pair, similar to how `HoodieInternalRowUtils.getCachedSchema` is used here?
<sub><i>- AI-generated; verify before applying. React 👍/👎 to flag quality.</i></sub>
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieSparkSqlWriter.scala:
##########
@@ -510,7 +510,9 @@ class HoodieSparkSqlWriterInternal {
val writeConfig = client.getConfig
if (writeConfig.getRecordMerger.getRecordType == HoodieRecordType.SPARK && tableType == MERGE_ON_READ && writeConfig.getLogDataBlockFormat.orElse(HoodieLogBlockType.AVRO_DATA_BLOCK) != HoodieLogBlockType.PARQUET_DATA_BLOCK) {
- throw new UnsupportedOperationException(s"${writeConfig.getRecordMerger.getClass.getName} only support parquet log.")
+ //TODO to revisit validity of this condition
+ // when commenting this out was able to get lance base files to work with avro log files
+ //throw new UnsupportedOperationException(s"${writeConfig.getRecordMerger.getClass.getName} only support parquet log.")
Review Comment:
🤖 nit: could you avoid shipping commented-out production code? It's hard for future readers to tell whether this guard was intentionally removed or is just temporarily bypassed — replacing it with a proper TODO comment (or a tracked issue reference) would make the intent much clearer.
<sub><i>- AI-generated; verify before applying. React 👍/👎 to flag quality.</i></sub>
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/execution/datasources/lance/SparkLanceReaderBase.scala:
##########
@@ -0,0 +1,183 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.spark.sql.execution.datasources.lance
+
+import org.apache.hudi.common.util
+import org.apache.hudi.internal.schema.InternalSchema
+import org.apache.hudi.io.storage.HoodieLanceRecordIterator
+import org.apache.hudi.storage.StorageConfiguration
+import com.lancedb.lance.file.LanceFileReader
+import org.apache.arrow.memory.RootAllocator
+import org.apache.hadoop.conf.Configuration
+import org.apache.spark.TaskContext
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions.{GenericInternalRow, JoinedRow}
+import org.apache.spark.sql.execution.datasources.{PartitionedFile, SparkColumnarFileReader}
+import org.apache.spark.sql.sources.Filter
+import org.apache.spark.sql.types.{StructField, StructType}
+import org.apache.spark.sql.util.LanceArrowUtils
+
+import java.io.IOException
+import scala.collection.JavaConverters._
+
+/**
+ * Reader for Lance files in Spark datasource.
+ * Implements vectorized reading using LanceArrowColumnVector.
+ *
+ * @param enableVectorizedReader whether to use vectorized reading (currently always true for Lance)
+ */
+class SparkLanceReaderBase(enableVectorizedReader: Boolean) extends SparkColumnarFileReader {
+
+ // Batch size for reading Lance files (number of rows per batch)
+ private val DEFAULT_BATCH_SIZE = 512
+
+ /**
+ * Read a Lance file with schema projection and partition column support.
+ *
+ * @param file Lance file to read
+ * @param requiredSchema desired output schema of the data (columns to read)
+ * @param partitionSchema schema of the partition columns. Partition values will be appended to the end of every row
+ * @param internalSchemaOpt option of internal schema for schema.on.read (not currently used for Lance)
+ * @param filters filters for data skipping. Not guaranteed to be used; the spark plan will also apply the filters.
+ * @param storageConf the hadoop conf
+ * @return iterator of rows read from the file output type says [[InternalRow]] but could be [[ColumnarBatch]]
+ */
+ override def read(file: PartitionedFile,
+ requiredSchema: StructType,
+ partitionSchema: StructType,
+ internalSchemaOpt: util.Option[InternalSchema],
+ filters: Seq[Filter],
+ storageConf: StorageConfiguration[Configuration]): Iterator[InternalRow] = {
+
+ val filePath = file.filePath.toString
+
+ if (requiredSchema.isEmpty && partitionSchema.isEmpty) {
+ // No columns requested - return empty iterator
+ Iterator.empty
+ } else {
+ // Create Arrow allocator for reading
+ val allocator = new RootAllocator(Long.MaxValue)
+
+ try {
+ // Open Lance file reader
+ val lanceReader = LanceFileReader.open(filePath, allocator)
+
+ // Get schema from Lance file
+ val arrowSchema = lanceReader.schema()
+ val fileSchema = LanceArrowUtils.fromArrowSchema(arrowSchema)
+
+ // TODO: Initial schema evolution (schema-on-read)
+ // Schema evolution: Filter to columns that currently exist in the file, as requested col may not be present
Review Comment:
🤖 Field-name comparison here is case-sensitive via `Set.contains`. Spark's normal schema resolution respects `spark.sql.caseSensitive`. If the table was written with `id` and a query references `ID` (case-insensitive mode), the field would be classified as 'missing' and the column would silently come back as NULL instead of populated. Is that the intended behavior, or should this honor the SQL conf?
<sub><i>- AI-generated; verify before applying. React 👍/👎 to flag quality.</i></sub>
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/execution/datasources/lance/SparkLanceReaderBase.scala:
##########
@@ -0,0 +1,183 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.spark.sql.execution.datasources.lance
+
+import org.apache.hudi.common.util
+import org.apache.hudi.internal.schema.InternalSchema
+import org.apache.hudi.io.storage.HoodieLanceRecordIterator
+import org.apache.hudi.storage.StorageConfiguration
+import com.lancedb.lance.file.LanceFileReader
+import org.apache.arrow.memory.RootAllocator
+import org.apache.hadoop.conf.Configuration
+import org.apache.spark.TaskContext
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions.{GenericInternalRow, JoinedRow}
+import org.apache.spark.sql.execution.datasources.{PartitionedFile, SparkColumnarFileReader}
+import org.apache.spark.sql.sources.Filter
+import org.apache.spark.sql.types.{StructField, StructType}
+import org.apache.spark.sql.util.LanceArrowUtils
+
+import java.io.IOException
+import scala.collection.JavaConverters._
+
+/**
+ * Reader for Lance files in Spark datasource.
+ * Implements vectorized reading using LanceArrowColumnVector.
+ *
+ * @param enableVectorizedReader whether to use vectorized reading (currently always true for Lance)
+ */
+class SparkLanceReaderBase(enableVectorizedReader: Boolean) extends SparkColumnarFileReader {
+
+ // Batch size for reading Lance files (number of rows per batch)
+ private val DEFAULT_BATCH_SIZE = 512
+
+ /**
+ * Read a Lance file with schema projection and partition column support.
+ *
+ * @param file Lance file to read
+ * @param requiredSchema desired output schema of the data (columns to read)
+ * @param partitionSchema schema of the partition columns. Partition values will be appended to the end of every row
+ * @param internalSchemaOpt option of internal schema for schema.on.read (not currently used for Lance)
+ * @param filters filters for data skipping. Not guaranteed to be used; the spark plan will also apply the filters.
+ * @param storageConf the hadoop conf
+ * @return iterator of rows read from the file output type says [[InternalRow]] but could be [[ColumnarBatch]]
+ */
+ override def read(file: PartitionedFile,
+ requiredSchema: StructType,
+ partitionSchema: StructType,
+ internalSchemaOpt: util.Option[InternalSchema],
+ filters: Seq[Filter],
+ storageConf: StorageConfiguration[Configuration]): Iterator[InternalRow] = {
+
+ val filePath = file.filePath.toString
+
+ if (requiredSchema.isEmpty && partitionSchema.isEmpty) {
+ // No columns requested - return empty iterator
+ Iterator.empty
+ } else {
+ // Create Arrow allocator for reading
+ val allocator = new RootAllocator(Long.MaxValue)
+
+ try {
+ // Open Lance file reader
+ val lanceReader = LanceFileReader.open(filePath, allocator)
+
+ // Get schema from Lance file
+ val arrowSchema = lanceReader.schema()
+ val fileSchema = LanceArrowUtils.fromArrowSchema(arrowSchema)
+
+ // TODO: Initial schema evolution (schema-on-read)
+ // Schema evolution: Filter to columns that currently exist in the file, as requested col may not be present
+ val fileFieldNames = fileSchema.fieldNames.toSet
+ val (existingFields, missingFields) = if (requiredSchema.nonEmpty) {
+ requiredSchema.fields.partition(f => fileFieldNames.contains(f.name))
+ } else {
+ (Array.empty[StructField], Array.empty[StructField])
+ }
+
+ // Only request columns that exist in the file
+ val columnNames: java.util.List[String] = if (existingFields.nonEmpty) {
+ existingFields.map(_.name).toList.asJava
+ } else {
+ // If only partition columns requested, read minimal data
Review Comment:
🤖 When `existingFields` is empty (e.g. count(*) with only partition cols, or a query that requests only newly-added columns that aren't in the file), `columnNames` becomes `null` and Lance reads ALL columns. The downstream iterator path in the `else` branch (no missing fields) then returns rows with all the file's columns rather than the empty/projected schema the caller asked for. Could you set `columnNames` to a single minimal column (or use a Lance API that returns row counts only) so the projection actually matches `requiredSchema`?
<sub><i>- AI-generated; verify before applying. React 👍/👎 to flag quality.</i></sub>
##########
hudi-common/src/main/java/org/apache/hudi/common/util/LanceUtils.java:
##########
@@ -0,0 +1,206 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.hudi.common.util;
+
+import org.apache.hudi.common.config.HoodieConfig;
+import org.apache.hudi.common.config.HoodieReaderConfig;
+import org.apache.hudi.common.model.HoodieFileFormat;
+import org.apache.hudi.common.model.HoodieKey;
+import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.common.util.collection.ClosableIterator;
+import org.apache.hudi.common.util.collection.CloseableMappingIterator;
+import org.apache.hudi.common.util.collection.Pair;
+import org.apache.hudi.exception.HoodieIOException;
+import org.apache.hudi.io.storage.HoodieFileReader;
+import org.apache.hudi.io.storage.HoodieIOFactory;
+import org.apache.hudi.keygen.BaseKeyGenerator;
+import org.apache.hudi.metadata.HoodieIndexVersion;
+import org.apache.hudi.stats.HoodieColumnRangeMetadata;
+import org.apache.hudi.storage.HoodieStorage;
+import org.apache.hudi.storage.StoragePath;
+
+import org.apache.avro.Schema;
+import org.apache.avro.generic.GenericRecord;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.Set;
+import java.util.concurrent.atomic.AtomicLong;
+
+public class LanceUtils extends FileFormatUtils {
+
+ @Override
+ public ClosableIterator<Pair<HoodieKey, Long>> fetchRecordKeysWithPositions(HoodieStorage storage, StoragePath filePath) {
+ return fetchRecordKeysWithPositions(storage, filePath, Option.empty(), Option.empty());
+ }
+
+ @Override
+ public ClosableIterator<Pair<HoodieKey, Long>> fetchRecordKeysWithPositions(HoodieStorage storage,
+ StoragePath filePath,
+ Option<BaseKeyGenerator> keyGeneratorOpt,
+ Option<String> partitionPath) {
+ AtomicLong position = new AtomicLong(0);
+ return new CloseableMappingIterator<>(
+ getHoodieKeyIterator(storage, filePath, keyGeneratorOpt, partitionPath),
+ key -> Pair.of(key, position.getAndIncrement()));
+ }
+
+ @Override
+ public ClosableIterator<HoodieKey> getHoodieKeyIterator(HoodieStorage storage, StoragePath filePath) {
+ return getHoodieKeyIterator(storage, filePath, Option.empty(), Option.empty());
+ }
+
+ @Override
+ public ClosableIterator<HoodieKey> getHoodieKeyIterator(HoodieStorage storage,
+ StoragePath filePath,
+ Option<BaseKeyGenerator> keyGeneratorOpt,
+ Option<String> partitionPath) {
+ try {
+ // Follow similar pattern as HFileUtils
+ HoodieFileReader reader = HoodieIOFactory.getIOFactory(storage)
+ .getReaderFactory(HoodieRecord.HoodieRecordType.SPARK)
+ .getFileReader(new HoodieReaderConfig(), filePath, HoodieFileFormat.LANCE);
+ ClosableIterator<String> keyIterator = reader.getRecordKeyIterator();
+ return new ClosableIterator<HoodieKey>() {
+ @Override
+ public void close() {
+ keyIterator.close();
+ }
+
+ @Override
+ public boolean hasNext() {
+ return keyIterator.hasNext();
+ }
+
+ @Override
+ public HoodieKey next() {
+ String key = keyIterator.next();
+ return new HoodieKey(key, partitionPath.orElse(null));
+ }
+ };
+ } catch (IOException e) {
+ throw new HoodieIOException("Failed to read from Lance file" + filePath, e);
+ }
+ }
+
+ @Override
+ public Schema readAvroSchema(HoodieStorage storage, StoragePath filePath) {
+ try (HoodieFileReader fileReader =
+ HoodieIOFactory.getIOFactory(storage)
+ .getReaderFactory(HoodieRecord.HoodieRecordType.SPARK)
+ .getFileReader(
+ ConfigUtils.DEFAULT_HUDI_CONFIG_FOR_READER,
+ filePath,
+ HoodieFileFormat.LANCE)) {
+ return fileReader.getSchema();
+ } catch (IOException e) {
+ throw new HoodieIOException("Failed to read schema from Lance file", e);
+ }
+ }
+
+ @Override
+ public HoodieFileFormat getFormat() {
+ return HoodieFileFormat.LANCE;
+ }
+
+ @Override
+ public List<GenericRecord> readAvroRecords(HoodieStorage storage, StoragePath filePath) {
+ throw new UnsupportedOperationException("readAvroRecords is not yet supported for Lance format");
+ }
+
+ @Override
+ public List<GenericRecord> readAvroRecords(HoodieStorage storage, StoragePath filePath, Schema schema) {
+ throw new UnsupportedOperationException("readAvroRecords with schema is not yet supported for Lance format");
+ }
+
+ @Override
+ public Map<String, String> readFooter(HoodieStorage storage, boolean required, StoragePath filePath, String... footerNames) {
+ throw new UnsupportedOperationException("readFooter is not yet supported for Lance format");
+ }
+
+ @Override
+ public long getRowCount(HoodieStorage storage, StoragePath filePath) {
+ try (HoodieFileReader fileReader =
+ HoodieIOFactory.getIOFactory(storage)
+ .getReaderFactory(HoodieRecord.HoodieRecordType.SPARK)
+ .getFileReader(
+ ConfigUtils.DEFAULT_HUDI_CONFIG_FOR_READER,
+ filePath,
+ HoodieFileFormat.LANCE)) {
+ return fileReader.getTotalRecords();
+ } catch (IOException e) {
+ throw new HoodieIOException("Failed to read schema from Lance file", e);
Review Comment:
🤖 nit: this error message reads "Failed to read schema from Lance file" but it's inside `getRowCount` — looks like a copy-paste from `readAvroSchema`. Could you update it to something like "Failed to get row count from Lance file" so it's actually useful when debugging?
<sub><i>- AI-generated; verify before applying. React 👍/👎 to flag quality.</i></sub>
##########
hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/io/storage/HoodieSparkLanceWriter.java:
##########
@@ -0,0 +1,147 @@
+/*
+ * 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.spark.arrow.LanceArrowWriter;
+import org.apache.arrow.vector.types.pojo.Schema;
+import org.apache.hudi.common.engine.TaskContextSupplier;
+import org.apache.hudi.common.model.HoodieKey;
+import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.io.lance.HoodieBaseLanceWriter;
+import org.apache.hudi.storage.HoodieStorage;
+import org.apache.hudi.storage.StoragePath;
+import org.apache.spark.sql.catalyst.InternalRow;
+import org.apache.spark.sql.types.StructType;
+import org.apache.spark.sql.util.LanceArrowUtils;
+import org.apache.spark.unsafe.types.UTF8String;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.function.Function;
+
+import static org.apache.hudi.common.model.HoodieRecord.HoodieMetadataField.COMMIT_SEQNO_METADATA_FIELD;
+import static org.apache.hudi.common.model.HoodieRecord.HoodieMetadataField.COMMIT_TIME_METADATA_FIELD;
+import static org.apache.hudi.common.model.HoodieRecord.HoodieMetadataField.FILENAME_METADATA_FIELD;
+import static org.apache.hudi.common.model.HoodieRecord.HoodieMetadataField.PARTITION_PATH_METADATA_FIELD;
+import static org.apache.hudi.common.model.HoodieRecord.HoodieMetadataField.RECORD_KEY_METADATA_FIELD;
+
+/**
+ * Spark Lance file writer implementing {@link HoodieSparkFileWriter}.
+ *
+ * This writer integrates with Hudi's storage I/O layer and supports:
+ * - Hudi metadata field population
+ * - Record key tracking (for bloom filters - TODO)
+ * - Sequence ID generation
+ */
+public class HoodieSparkLanceWriter extends HoodieBaseLanceWriter<InternalRow> implements HoodieSparkFileWriter {
+
+ private static final long DEFAULT_MAX_FILE_SIZE = 120 * 1024 * 1024; // 120MB
+ private static final String DEFAULT_TIMEZONE = "UTC";
Review Comment:
🤖 nit: `DEFAULT_MAX_FILE_SIZE` and `DEFAULT_TIMEZONE` are defined identically in both `HoodieSparkLanceWriter` and `HoodieInternalRowLanceWriter` — could these be promoted to `HoodieBaseLanceWriter` to avoid the duplication? If one gets updated and the other is missed, the two writers would silently diverge.
<sub><i>- AI-generated; verify before applying. React 👍/👎 to flag quality.</i></sub>
--
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