FrankChen021 commented on code in PR #19510: URL: https://github.com/apache/druid/pull/19510#discussion_r3979574856
########## extensions-contrib/druid-iceberg-extensions/src/main/java/org/apache/druid/iceberg/input/IcebergArrowInputSourceReader.java: ########## @@ -0,0 +1,381 @@ +/* + * 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.druid.iceberg.input; + +import com.google.common.collect.Maps; +import org.apache.arrow.vector.FieldVector; +import org.apache.druid.data.input.ColumnsFilter; +import org.apache.druid.data.input.InputRow; +import org.apache.druid.data.input.InputRowListPlusRawValues; +import org.apache.druid.data.input.InputRowSchema; +import org.apache.druid.data.input.InputSourceReader; +import org.apache.druid.data.input.InputStats; +import org.apache.druid.data.input.MapBasedInputRow; +import org.apache.druid.error.DruidException; +import org.apache.druid.iceberg.filter.IcebergFilter; +import org.apache.druid.java.util.common.parsers.CloseableIterator; +import org.apache.iceberg.CombinedScanTask; +import org.apache.iceberg.FileScanTask; +import org.apache.iceberg.Schema; +import org.apache.iceberg.Table; +import org.apache.iceberg.TableScan; +import org.apache.iceberg.arrow.vectorized.ArrowReader; +import org.apache.iceberg.arrow.vectorized.ColumnVector; +import org.apache.iceberg.arrow.vectorized.ColumnarBatch; +import org.apache.iceberg.io.CloseableIterable; +import org.apache.iceberg.types.Type; +import org.apache.iceberg.types.Types; +import org.apache.iceberg.util.TableScanUtil; +import org.joda.time.DateTime; + +import javax.annotation.Nullable; +import java.io.IOException; +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import java.util.NoSuchElementException; +import java.util.concurrent.TimeUnit; +import java.util.stream.Collectors; + +/** + * Reads an Iceberg table via iceberg-arrow's {@link ArrowReader}, yielding {@link InputRow} objects. + * + * Type coercion and schema evolution are handled entirely by the Iceberg library. Druid only consumes + * the resulting {@link ColumnarBatch} batches and maps them to {@link MapBasedInputRow}. + * + * Column projection and predicate push-down are applied at scan planning time so only requested + * columns and matching files are read from storage. + * + * Note: iceberg-arrow currently supports Parquet data files only. ORC and Avro files will throw + * {@link UnsupportedOperationException} at read time. Delete-file snapshots are rejected because + * iceberg-arrow does not apply equality or positional deletes. + */ +public class IcebergArrowInputSourceReader implements InputSourceReader +{ + static final int DEFAULT_BATCH_SIZE = 1024; + + private final Table table; + @Nullable + private final IcebergFilter icebergFilter; + @Nullable + private final DateTime snapshotTime; + private final boolean caseSensitive; + private final InputRowSchema schema; + private final int batchSize; + + public IcebergArrowInputSourceReader( + final Table table, + @Nullable final IcebergFilter icebergFilter, + @Nullable final DateTime snapshotTime, + final boolean caseSensitive, + final InputRowSchema schema, + final int batchSize + ) + { + this.table = table; + this.icebergFilter = icebergFilter; + this.snapshotTime = snapshotTime; + this.caseSensitive = caseSensitive; + this.schema = schema; + this.batchSize = batchSize; + } + + @Override + public CloseableIterator<InputRow> read(@Nullable final InputStats inputStats) throws IOException + { + final TableScan scan = buildScan(); + validateNoDeleteFiles(scan); + final CloseableIterable<CombinedScanTask> tasks = TableScanUtil.planTasks( + scan.planFiles(), + scan.targetSplitSize(), + scan.splitLookback(), + scan.splitOpenFileCost() + ); + final ArrowReader arrowReader = new ArrowReader(scan, batchSize, true); + final org.apache.iceberg.io.CloseableIterator<ColumnarBatch> batchIter = arrowReader.open(tasks); Review Comment: [P1] Keep the extension classloader for Arrow iteration retrieveTable() restores the thread context classloader before this Arrow reader is opened. In Iceberg 1.11, ArrowReader lazily asks FormatModelRegistry to reflectively load org.apache.iceberg.arrow.vectorized.ArrowFormatModels through the thread context loader as the iterator advances. In a normal installed Druid extension that loader does not contain iceberg-arrow, so the first batch can fail to find/register the Parquet ColumnarBatch model even though classpath-based tests pass. Keep getClass().getClassLoader() active across Arrow reader creation and the returned iterator lifecycle, or explicitly register the model with the extension loader. ########## extensions-contrib/druid-iceberg-extensions/src/main/java/org/apache/druid/iceberg/input/IcebergInputSource.java: ########## @@ -125,97 +195,248 @@ public Stream<InputSplit<List<String>>> createSplits( @Nullable SplitHintSpec splitHintSpec ) throws IOException { - if (!isLoaded) { - retrieveIcebergDatafiles(); - } - return getDelegateInputSource().createSplits(inputFormat, splitHintSpec); + return delegate.createSplits(inputFormat, splitHintSpec); } @Override public int estimateNumSplits(InputFormat inputFormat, @Nullable SplitHintSpec splitHintSpec) throws IOException { - if (!isLoaded) { - retrieveIcebergDatafiles(); - } - return getDelegateInputSource().estimateNumSplits(inputFormat, splitHintSpec); + return delegate.estimateNumSplits(inputFormat, splitHintSpec); } @Override public InputSource withSplit(InputSplit<List<String>> inputSplit) { - return getDelegateInputSource().withSplit(inputSplit); + return delegate.withSplit(inputSplit); } @Override public SplitHintSpec getSplitHintSpecOrDefault(@Nullable SplitHintSpec splitHintSpec) { - return getDelegateInputSource().getSplitHintSpecOrDefault(splitHintSpec); + return delegate.getSplitHintSpecOrDefault(splitHintSpec); } - @JsonProperty - public String getTableName() + private Table retrieveTable() { - return tableName; + return icebergCatalog.retrieveTable(namespace, tableName); } - @JsonProperty - public String getNamespace() + /** + * Mode-specific behavior. The two modes differ on more than how rows are read: they disagree on whether + * an {@link InputFormat} is needed and whether the source can be split across tasks. + */ + private interface InputSourceDelegate { - return namespace; - } + boolean needsFormat(); - @JsonProperty - public IcebergCatalog getIcebergCatalog() - { - return icebergCatalog; - } + boolean isSplittable(); - @JsonProperty - public IcebergFilter getIcebergFilter() - { - return icebergFilter; - } + InputSourceReader reader( + InputRowSchema inputRowSchema, + @Nullable InputFormat inputFormat, + File temporaryDirectory + ); - @Nullable - @JsonProperty - public DateTime getSnapshotTime() - { - return snapshotTime; - } + Stream<InputSplit<List<String>>> createSplits( + InputFormat inputFormat, + @Nullable SplitHintSpec splitHintSpec + ) throws IOException; - @JsonProperty - public ResidualFilterMode getResidualFilterMode() - { - return residualFilterMode; - } + int estimateNumSplits(InputFormat inputFormat, @Nullable SplitHintSpec splitHintSpec) throws IOException; - public SplittableInputSource getDelegateInputSource() - { - return delegateInputSource; + InputSource withSplit(InputSplit<List<String>> inputSplit); + + SplitHintSpec getSplitHintSpecOrDefault(@Nullable SplitHintSpec splitHintSpec); } - protected void retrieveIcebergDatafiles() + /** + * Resolves the snapshot to a list of data file paths and defers reading to the warehouse input source. + */ + private class StandardDelegate implements InputSourceDelegate { - List<String> snapshotDataFiles = icebergCatalog.extractSnapshotDataFiles( - getNamespace(), - getTableName(), - getIcebergFilter(), - getSnapshotTime(), - getResidualFilterMode() - ); - if (snapshotDataFiles.isEmpty()) { - delegateInputSource = new EmptyInputSource(); - } else { - delegateInputSource = warehouseSource.create(snapshotDataFiles); + private final InputSourceFactory warehouseSource; + + private boolean isLoaded = false; + private SplittableInputSource delegateInputSource; + + StandardDelegate(final InputSourceFactory warehouseSource) + { + this.warehouseSource = warehouseSource; + } + + @Override + public boolean needsFormat() + { + return true; + } + + @Override + public boolean isSplittable() + { + return true; + } + + @Override + public InputSourceReader reader( + InputRowSchema inputRowSchema, + @Nullable InputFormat inputFormat, + File temporaryDirectory + ) + { + return warehouseInputSource().reader(inputRowSchema, inputFormat, temporaryDirectory); + } + + @Override + public Stream<InputSplit<List<String>>> createSplits( + InputFormat inputFormat, + @Nullable SplitHintSpec splitHintSpec + ) throws IOException + { + return warehouseInputSource().createSplits(inputFormat, splitHintSpec); + } + + @Override + public int estimateNumSplits(InputFormat inputFormat, @Nullable SplitHintSpec splitHintSpec) throws IOException + { + return warehouseInputSource().estimateNumSplits(inputFormat, splitHintSpec); + } + + @Override + public InputSource withSplit(InputSplit<List<String>> inputSplit) + { + return warehouseInputSource().withSplit(inputSplit); + } + + @Override + public SplitHintSpec getSplitHintSpecOrDefault(@Nullable SplitHintSpec splitHintSpec) + { + return warehouseInputSource().getSplitHintSpecOrDefault(splitHintSpec); + } + + private SplittableInputSource warehouseInputSource() + { + if (!isLoaded) { + final List<String> snapshotDataFiles = icebergCatalog.extractSnapshotDataFiles( + getNamespace(), + getTableName(), + getIcebergFilter(), + getSnapshotTime(), + getResidualFilterMode() + ); + if (snapshotDataFiles.isEmpty()) { + delegateInputSource = new EmptyInputSource(); + } else { + delegateInputSource = warehouseSource.create(snapshotDataFiles); + } + isLoaded = true; + } + return delegateInputSource; } - isLoaded = true; } /** - * This input source is used in place of a delegate input source if there are no input file paths. - * Certain input sources cannot be instantiated with an empty input file list and so composing input sources such as IcebergInputSource - * may use this input source as delegate in such cases. + * Scans the table through Iceberg's vectorized Arrow reader. Parquet only, and not splittable: + * there is no data file list to hand out, so all rows are read by a single task. */ + private class ArrowDelegate implements InputSourceDelegate + { + @Override + public boolean needsFormat() + { + return false; + } + + @Override + public boolean isSplittable() + { + return false; + } + + @Override + public InputSourceReader reader( + InputRowSchema inputRowSchema, + @Nullable InputFormat inputFormat, + File temporaryDirectory + ) + { + final Table table = retrieveTable(); + TableScan scan = table.newScan().caseSensitive(icebergCatalog.isCaseSensitive()); + if (icebergFilter != null) { + scan = icebergFilter.filter(scan); + } + if (snapshotTime != null) { + scan = scan.asOfTime(snapshotTime.getMillis()); + } + if (icebergFilter != null) { + icebergCatalog.enforceResidualMode(scan, residualFilterMode); Review Comment: [P1] Preserve IGNORE residual semantics This only validates the residual mode; the reader below still builds the scan with the original filter. Iceberg 1.11's ArrowReader passes each FileScanTask.residual() to Parquet's record filter, so an unpartitioned equals filter removes non-matching rows. That contradicts this source's documented/default IGNORE behavior, where residual rows are ingested and may be filtered by Druid's transformSpec, and it also conflicts with the new test expecting all three rows. Disable the record filter for IGNORE (while retaining it for an explicitly supported mode), or change the mode's contract and tests. ########## extensions-contrib/druid-iceberg-extensions/src/main/java/org/apache/druid/iceberg/input/IcebergArrowInputSourceReader.java: ########## @@ -0,0 +1,381 @@ +/* + * 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.druid.iceberg.input; + +import com.google.common.collect.Maps; +import org.apache.arrow.vector.FieldVector; +import org.apache.druid.data.input.ColumnsFilter; +import org.apache.druid.data.input.InputRow; +import org.apache.druid.data.input.InputRowListPlusRawValues; +import org.apache.druid.data.input.InputRowSchema; +import org.apache.druid.data.input.InputSourceReader; +import org.apache.druid.data.input.InputStats; +import org.apache.druid.data.input.MapBasedInputRow; +import org.apache.druid.error.DruidException; +import org.apache.druid.iceberg.filter.IcebergFilter; +import org.apache.druid.java.util.common.parsers.CloseableIterator; +import org.apache.iceberg.CombinedScanTask; +import org.apache.iceberg.FileScanTask; +import org.apache.iceberg.Schema; +import org.apache.iceberg.Table; +import org.apache.iceberg.TableScan; +import org.apache.iceberg.arrow.vectorized.ArrowReader; +import org.apache.iceberg.arrow.vectorized.ColumnVector; +import org.apache.iceberg.arrow.vectorized.ColumnarBatch; +import org.apache.iceberg.io.CloseableIterable; +import org.apache.iceberg.types.Type; +import org.apache.iceberg.types.Types; +import org.apache.iceberg.util.TableScanUtil; +import org.joda.time.DateTime; + +import javax.annotation.Nullable; +import java.io.IOException; +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import java.util.NoSuchElementException; +import java.util.concurrent.TimeUnit; +import java.util.stream.Collectors; + +/** + * Reads an Iceberg table via iceberg-arrow's {@link ArrowReader}, yielding {@link InputRow} objects. + * + * Type coercion and schema evolution are handled entirely by the Iceberg library. Druid only consumes + * the resulting {@link ColumnarBatch} batches and maps them to {@link MapBasedInputRow}. + * + * Column projection and predicate push-down are applied at scan planning time so only requested + * columns and matching files are read from storage. + * + * Note: iceberg-arrow currently supports Parquet data files only. ORC and Avro files will throw + * {@link UnsupportedOperationException} at read time. Delete-file snapshots are rejected because + * iceberg-arrow does not apply equality or positional deletes. + */ +public class IcebergArrowInputSourceReader implements InputSourceReader +{ + static final int DEFAULT_BATCH_SIZE = 1024; + + private final Table table; + @Nullable + private final IcebergFilter icebergFilter; + @Nullable + private final DateTime snapshotTime; + private final boolean caseSensitive; + private final InputRowSchema schema; + private final int batchSize; + + public IcebergArrowInputSourceReader( + final Table table, + @Nullable final IcebergFilter icebergFilter, + @Nullable final DateTime snapshotTime, + final boolean caseSensitive, + final InputRowSchema schema, + final int batchSize + ) + { + this.table = table; + this.icebergFilter = icebergFilter; + this.snapshotTime = snapshotTime; + this.caseSensitive = caseSensitive; + this.schema = schema; + this.batchSize = batchSize; + } + + @Override + public CloseableIterator<InputRow> read(@Nullable final InputStats inputStats) throws IOException + { + final TableScan scan = buildScan(); + validateNoDeleteFiles(scan); + final CloseableIterable<CombinedScanTask> tasks = TableScanUtil.planTasks( + scan.planFiles(), + scan.targetSplitSize(), + scan.splitLookback(), + scan.splitOpenFileCost() + ); + final ArrowReader arrowReader = new ArrowReader(scan, batchSize, true); + final org.apache.iceberg.io.CloseableIterator<ColumnarBatch> batchIter = arrowReader.open(tasks); + return new ArrowInputRowIterator( + batchIter, + arrowReader, + tasks, + inputStats != null ? inputStats : new NoopInputStats(), + scan.schema() + ); + } + + private void validateNoDeleteFiles(final TableScan scan) throws IOException + { + try (CloseableIterable<FileScanTask> fileTasks = scan.planFiles()) { + for (FileScanTask fileTask : fileTasks) { + if (!fileTask.deletes().isEmpty()) { + throw DruidException.forPersona(DruidException.Persona.USER) + .ofCategory(DruidException.Category.UNSUPPORTED) + .build( + "Arrow reader does not support Iceberg snapshots with delete files. " + + "Use a delete-aware input path." + ); + } + } + } + } + + @Override + public CloseableIterator<InputRowListPlusRawValues> sample() throws IOException + { + final CloseableIterator<InputRow> rows = read(new NoopInputStats()); + return new CloseableIterator<InputRowListPlusRawValues>() + { + @Override + public boolean hasNext() + { + return rows.hasNext(); + } + + @Override + public InputRowListPlusRawValues next() + { + final InputRow row = rows.next(); + return InputRowListPlusRawValues.of(row, ((MapBasedInputRow) row).getEvent()); + } + + @Override + public void close() throws IOException + { + rows.close(); + } + }; + } + + private TableScan buildScan() + { + TableScan scan = table.newScan().caseSensitive(caseSensitive); + + if (snapshotTime != null) { + scan = scan.asOfTime(snapshotTime.getMillis()); + } + + final List<String> projection = projectedColumns(scan.schema()); + if (projection != null) { + scan = scan.select(projection); + } + if (icebergFilter != null) { + scan = icebergFilter.filter(scan); + } + return scan; + } + + /** Projection authority is ColumnsFilter, not DimensionsSpec. Mirrors DeltaInputSource#pruneSchema. */ + @Nullable + private List<String> projectedColumns(final Schema scanSchema) + { + final ColumnsFilter filter = schema.getColumnsFilter(); + final List<String> allColumns = scanSchema.columns().stream() + .map(Types.NestedField::name) + .collect(Collectors.toList()); + final List<String> filtered = allColumns.stream() + .filter(filter::apply) + .collect(Collectors.toList()); + if (filtered.equals(allColumns)) { + return null; + } + final String tsCol = schema.getTimestampSpec().getTimestampColumn(); + if (tsCol != null && allColumns.contains(tsCol) && !filtered.contains(tsCol)) { + filtered.add(tsCol); + } + return filtered; + } + + private InputRow batchRowToInputRow( + final ColumnarBatch batch, + final int rowIdx, + final Schema readSchema + ) + { + final int numCols = batch.numCols(); + final Map<String, Object> event = Maps.newHashMapWithExpectedSize(numCols); + for (int col = 0; col < numCols; col++) { + final ColumnVector column = batch.column(col); + final FieldVector vec = column.getFieldVector(); + if (!column.isNullAt(rowIdx)) { + event.put(vec.getName(), extractValue(column, readSchema.findField(vec.getName()).type(), rowIdx)); Review Comment: [P1] Guard against type promotion failures readSchema is the logical table schema, but iceberg-arrow 1.11 allocates vectors from each Parquet file's physical/logical type and does not handle every legal promotion. For an existing int field promoted to long, an old Parquet INT32/INT(32) column can make ArrowReader cast a BigIntVector to IntVector while allocating the batch, before this accessor can read it; float-to-double has the same shape. A normal Iceberg type evolution therefore makes Arrow ingestion fail on the first old file. Detect promoted fields and use the standard reader/fail clearly, and do not claim schema evolution is fully delegated to Iceberg without covering this case. ########## extensions-contrib/druid-iceberg-extensions/src/main/java/org/apache/druid/iceberg/input/IcebergArrowInputSourceReader.java: ########## @@ -0,0 +1,381 @@ +/* + * 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.druid.iceberg.input; + +import com.google.common.collect.Maps; +import org.apache.arrow.vector.FieldVector; +import org.apache.druid.data.input.ColumnsFilter; +import org.apache.druid.data.input.InputRow; +import org.apache.druid.data.input.InputRowListPlusRawValues; +import org.apache.druid.data.input.InputRowSchema; +import org.apache.druid.data.input.InputSourceReader; +import org.apache.druid.data.input.InputStats; +import org.apache.druid.data.input.MapBasedInputRow; +import org.apache.druid.error.DruidException; +import org.apache.druid.iceberg.filter.IcebergFilter; +import org.apache.druid.java.util.common.parsers.CloseableIterator; +import org.apache.iceberg.CombinedScanTask; +import org.apache.iceberg.FileScanTask; +import org.apache.iceberg.Schema; +import org.apache.iceberg.Table; +import org.apache.iceberg.TableScan; +import org.apache.iceberg.arrow.vectorized.ArrowReader; +import org.apache.iceberg.arrow.vectorized.ColumnVector; +import org.apache.iceberg.arrow.vectorized.ColumnarBatch; +import org.apache.iceberg.io.CloseableIterable; +import org.apache.iceberg.types.Type; +import org.apache.iceberg.types.Types; +import org.apache.iceberg.util.TableScanUtil; +import org.joda.time.DateTime; + +import javax.annotation.Nullable; +import java.io.IOException; +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import java.util.NoSuchElementException; +import java.util.concurrent.TimeUnit; +import java.util.stream.Collectors; + +/** + * Reads an Iceberg table via iceberg-arrow's {@link ArrowReader}, yielding {@link InputRow} objects. + * + * Type coercion and schema evolution are handled entirely by the Iceberg library. Druid only consumes + * the resulting {@link ColumnarBatch} batches and maps them to {@link MapBasedInputRow}. + * + * Column projection and predicate push-down are applied at scan planning time so only requested + * columns and matching files are read from storage. + * + * Note: iceberg-arrow currently supports Parquet data files only. ORC and Avro files will throw + * {@link UnsupportedOperationException} at read time. Delete-file snapshots are rejected because + * iceberg-arrow does not apply equality or positional deletes. + */ +public class IcebergArrowInputSourceReader implements InputSourceReader +{ + static final int DEFAULT_BATCH_SIZE = 1024; + + private final Table table; + @Nullable + private final IcebergFilter icebergFilter; + @Nullable + private final DateTime snapshotTime; + private final boolean caseSensitive; + private final InputRowSchema schema; + private final int batchSize; + + public IcebergArrowInputSourceReader( + final Table table, + @Nullable final IcebergFilter icebergFilter, + @Nullable final DateTime snapshotTime, + final boolean caseSensitive, + final InputRowSchema schema, + final int batchSize + ) + { + this.table = table; + this.icebergFilter = icebergFilter; + this.snapshotTime = snapshotTime; + this.caseSensitive = caseSensitive; + this.schema = schema; + this.batchSize = batchSize; + } + + @Override + public CloseableIterator<InputRow> read(@Nullable final InputStats inputStats) throws IOException + { + final TableScan scan = buildScan(); + validateNoDeleteFiles(scan); + final CloseableIterable<CombinedScanTask> tasks = TableScanUtil.planTasks( + scan.planFiles(), + scan.targetSplitSize(), + scan.splitLookback(), + scan.splitOpenFileCost() + ); + final ArrowReader arrowReader = new ArrowReader(scan, batchSize, true); + final org.apache.iceberg.io.CloseableIterator<ColumnarBatch> batchIter = arrowReader.open(tasks); + return new ArrowInputRowIterator( + batchIter, + arrowReader, + tasks, + inputStats != null ? inputStats : new NoopInputStats(), + scan.schema() + ); + } + + private void validateNoDeleteFiles(final TableScan scan) throws IOException + { + try (CloseableIterable<FileScanTask> fileTasks = scan.planFiles()) { + for (FileScanTask fileTask : fileTasks) { + if (!fileTask.deletes().isEmpty()) { + throw DruidException.forPersona(DruidException.Persona.USER) + .ofCategory(DruidException.Category.UNSUPPORTED) + .build( + "Arrow reader does not support Iceberg snapshots with delete files. " + + "Use a delete-aware input path." + ); + } + } + } + } + + @Override + public CloseableIterator<InputRowListPlusRawValues> sample() throws IOException + { + final CloseableIterator<InputRow> rows = read(new NoopInputStats()); + return new CloseableIterator<InputRowListPlusRawValues>() + { + @Override + public boolean hasNext() + { + return rows.hasNext(); + } + + @Override + public InputRowListPlusRawValues next() + { + final InputRow row = rows.next(); + return InputRowListPlusRawValues.of(row, ((MapBasedInputRow) row).getEvent()); + } + + @Override + public void close() throws IOException + { + rows.close(); + } + }; + } + + private TableScan buildScan() + { + TableScan scan = table.newScan().caseSensitive(caseSensitive); + + if (snapshotTime != null) { + scan = scan.asOfTime(snapshotTime.getMillis()); + } + + final List<String> projection = projectedColumns(scan.schema()); + if (projection != null) { + scan = scan.select(projection); + } + if (icebergFilter != null) { + scan = icebergFilter.filter(scan); + } + return scan; + } + + /** Projection authority is ColumnsFilter, not DimensionsSpec. Mirrors DeltaInputSource#pruneSchema. */ + @Nullable + private List<String> projectedColumns(final Schema scanSchema) + { + final ColumnsFilter filter = schema.getColumnsFilter(); + final List<String> allColumns = scanSchema.columns().stream() + .map(Types.NestedField::name) + .collect(Collectors.toList()); + final List<String> filtered = allColumns.stream() + .filter(filter::apply) + .collect(Collectors.toList()); + if (filtered.equals(allColumns)) { + return null; + } + final String tsCol = schema.getTimestampSpec().getTimestampColumn(); + if (tsCol != null && allColumns.contains(tsCol) && !filtered.contains(tsCol)) { + filtered.add(tsCol); + } + return filtered; + } + + private InputRow batchRowToInputRow( + final ColumnarBatch batch, + final int rowIdx, + final Schema readSchema + ) + { + final int numCols = batch.numCols(); + final Map<String, Object> event = Maps.newHashMapWithExpectedSize(numCols); + for (int col = 0; col < numCols; col++) { + final ColumnVector column = batch.column(col); + final FieldVector vec = column.getFieldVector(); Review Comment: [P1] Handle fields absent from older files When an optional Iceberg field is added after an existing Parquet file, the per-file Arrow reader uses a dummy/null holder for that field. The holder has no FieldVector, but this line dereferences it before checking column.isNullAt; resolveDimensions() and estimateBatchBytes() make the same assumption. Reading a mixed old/new table therefore fails instead of producing null for the old rows (and Iceberg 1.11 may fail while constructing the batch for the dummy holder). Derive names from readSchema and handle dummy/constant columns, or reject/fallback for this schema evolution. -- 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] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
