FrankChen021 commented on code in PR #19510:
URL: https://github.com/apache/druid/pull/19510#discussion_r3989887484
##########
extensions-contrib/druid-iceberg-extensions/src/main/java/org/apache/druid/iceberg/input/IcebergInputSource.java:
##########
@@ -34,76 +34,149 @@
import org.apache.druid.data.input.InputStats;
import org.apache.druid.data.input.SplitHintSpec;
import org.apache.druid.data.input.impl.SplittableInputSource;
+import org.apache.druid.error.DruidException;
import org.apache.druid.iceberg.filter.IcebergFilter;
import org.apache.druid.java.util.common.CloseableIterators;
import org.apache.druid.java.util.common.parsers.CloseableIterator;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.FileScanTask;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.TableScan;
+import org.apache.iceberg.io.CloseableIterable;
import org.joda.time.DateTime;
import javax.annotation.Nullable;
import java.io.File;
import java.io.IOException;
+import java.io.UncheckedIOException;
import java.util.Collections;
import java.util.List;
import java.util.stream.Stream;
/**
- * Inputsource to ingest data managed by the Iceberg table format.
- * This inputsource talks to the configured catalog, executes any configured
filters and retrieves the data file paths upto the latest snapshot associated
with the iceberg table.
- * The data file paths are then provided to a native {@link
SplittableInputSource} implementation depending on the warehouse source defined.
+ * Reads an Iceberg table. Two reader modes sit behind this single type:
+ * the default resolves the snapshot to data file paths and hands them to
{@code warehouseSource},
+ * while {@code useArrowReader} scans the table directly through Iceberg's
vectorized Arrow reader.
*/
public class IcebergInputSource implements SplittableInputSource<List<String>>
{
public static final String TYPE_KEY = "iceberg";
- @JsonProperty
private final String tableName;
-
- @JsonProperty
private final String namespace;
-
- @JsonProperty
- private IcebergCatalog icebergCatalog;
-
- @JsonProperty
- private IcebergFilter icebergFilter;
-
- @JsonProperty
- private InputSourceFactory warehouseSource;
-
- @JsonProperty
+ private final IcebergCatalog icebergCatalog;
+ private final IcebergFilter icebergFilter;
private final DateTime snapshotTime;
-
- @JsonProperty
private final ResidualFilterMode residualFilterMode;
+ private final boolean useArrowReader;
+ private final int arrowBatchSize;
- private boolean isLoaded = false;
+ @Nullable
+ private final InputSourceFactory warehouseSource;
- private SplittableInputSource delegateInputSource;
+ private final InputSourceDelegate delegate;
@JsonCreator
public IcebergInputSource(
@JsonProperty("tableName") String tableName,
@JsonProperty("namespace") String namespace,
@JsonProperty("icebergFilter") @Nullable IcebergFilter icebergFilter,
@JsonProperty("icebergCatalog") IcebergCatalog icebergCatalog,
- @JsonProperty("warehouseSource") InputSourceFactory warehouseSource,
+ @JsonProperty("warehouseSource") @Nullable InputSourceFactory
warehouseSource,
@JsonProperty("snapshotTime") @Nullable DateTime snapshotTime,
- @JsonProperty("residualFilterMode") @Nullable ResidualFilterMode
residualFilterMode
+ @JsonProperty("residualFilterMode") @Nullable ResidualFilterMode
residualFilterMode,
+ @JsonProperty("useArrowReader") @Nullable Boolean useArrowReader,
Review Comment:
[P2] Preserve the existing Java construction API
This replaces the only public seven-argument `IcebergInputSource`
constructor with a nine-argument constructor, so existing Java callers and
already-compiled clients that construct the source no longer compile or link
even when Arrow is disabled. The refactor also removes the public
`getDelegateInputSource()` and protected `retrieveIcebergDatafiles()` hooks.
Keep a delegating seven-argument overload and preserve or deprecate the old
hooks, or explicitly stage this as a breaking API change.
##########
extensions-contrib/druid-iceberg-extensions/src/main/java/org/apache/druid/iceberg/input/IcebergArrowInputSourceReader.java:
##########
@@ -0,0 +1,432 @@
+/*
+ * 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 compatible schema evolution are handled 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 ClassLoader extensionClassLoader =
IcebergArrowInputSourceReader.class.getClassLoader();
+ final ClassLoader originalClassLoader =
Thread.currentThread().getContextClassLoader();
+ try {
+ Thread.currentThread().setContextClassLoader(extensionClassLoader);
+ 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(),
+ extensionClassLoader
+ );
+ }
+ finally {
+ Thread.currentThread().setContextClassLoader(originalClassLoader);
+ }
+ }
+
+ 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 Types.NestedField field = readSchema.columns().get(col);
+ if (!column.isNullAt(rowIdx)) {
+ event.put(field.name(), extractValue(column, field.type(), rowIdx));
+ }
+ }
+ final long timestamp =
schema.getTimestampSpec().extractTimestamp(event).getMillis();
+ final List<String> dimensions = resolveDimensions(readSchema);
+ return new MapBasedInputRow(timestamp, dimensions, event);
+ }
+
+ private List<String> resolveDimensions(final Schema readSchema)
+ {
+ final List<String> configured =
schema.getDimensionsSpec().getDimensionNames();
+ if (!configured.isEmpty()) {
+ return configured;
+ }
+ final String tsCol = schema.getTimestampSpec().getTimestampColumn();
+ final List<String> dims = new ArrayList<>(readSchema.columns().size());
+ for (final Types.NestedField field : readSchema.columns()) {
+ if (!field.name().equals(tsCol)) {
+ dims.add(field.name());
+ }
+ }
+ return dims;
+ }
+
+ /**
+ * Type-safe extraction from Iceberg column accessors so physical dictionary
encoding is not exposed.
+ * Covers all scalar types supported by iceberg-arrow 1.10.0.
+ */
+ static Object extractValue(final ColumnVector column, final Type type, final
int idx)
+ {
+ switch (type.typeId()) {
+ case BOOLEAN:
+ return column.getBoolean(idx);
+ case INTEGER:
+ return column.getInt(idx);
+ case LONG:
+ return column.getLong(idx);
+ case FLOAT:
+ return (double) column.getFloat(idx);
+ case DOUBLE:
+ return column.getDouble(idx);
+ case STRING:
+ return column.getString(idx);
+ case BINARY:
+ case FIXED:
+ case UUID:
+ return column.getBinary(idx);
+ case DATE:
+ return TimeUnit.DAYS.toMillis(column.getInt(idx));
+ case TIME:
+ return TimeUnit.MICROSECONDS.toMillis(column.getLong(idx));
+ case TIMESTAMP:
+ return TimeUnit.MICROSECONDS.toMillis(column.getLong(idx));
+ case TIMESTAMP_NANO:
+ return TimeUnit.NANOSECONDS.toMillis(column.getLong(idx));
+ case DECIMAL:
+ final Types.DecimalType decimalType = (Types.DecimalType) type;
+ return column.getDecimal(idx, decimalType.precision(),
decimalType.scale());
Review Comment:
[P1] Do not truncate high-precision decimals
`ColumnVector.getDecimal` in iceberg-arrow 1.11's Java decimal accessor
converts the decoded value with `value.unscaledValue().longValue()` before
constructing the returned `BigDecimal`. A valid DECIMAL(20,2) or
DECIMAL(38,...) value whose unscaled value exceeds `long` is therefore silently
truncated/wrapped before it reaches the Druid event. This path advertises
scalar support but neither limits decimal precision nor tests it; the result is
incorrect data rather than a clean unsupported-input failure. Preserve
arbitrary-precision values or reject unsupported decimal schemas/values before
selecting Arrow.
##########
extensions-contrib/druid-iceberg-extensions/src/main/java/org/apache/druid/iceberg/input/IcebergArrowInputSourceReader.java:
##########
@@ -0,0 +1,432 @@
+/*
+ * 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 compatible schema evolution are handled 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 ClassLoader extensionClassLoader =
IcebergArrowInputSourceReader.class.getClassLoader();
+ final ClassLoader originalClassLoader =
Thread.currentThread().getContextClassLoader();
+ try {
+ Thread.currentThread().setContextClassLoader(extensionClassLoader);
+ final ArrowReader arrowReader = new ArrowReader(scan, batchSize, true);
+ final org.apache.iceberg.io.CloseableIterator<ColumnarBatch> batchIter =
arrowReader.open(tasks);
Review Comment:
[P2] Close scan tasks when reader setup fails
`TableScanUtil.planTasks` returns a closeable iterable that owns the
underlying planned-file iterable, but it is only closed by
`ArrowInputRowIterator.close()`. If `arrowReader.open(tasks)` throws while
materializing an unsupported type, empty projection, bad delete state, or
another setup error, no iterator is returned and neither `tasks` nor the
partially-created reader is closed. Repeated failed Arrow ingestions can
consequently retain scan resources; close the task iterable on every
setup-failure path while transferring ownership only after `open` succeeds.
##########
extensions-contrib/druid-iceberg-extensions/src/main/java/org/apache/druid/iceberg/input/IcebergArrowInputSourceReader.java:
##########
@@ -0,0 +1,432 @@
+/*
+ * 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 compatible schema evolution are handled 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 ClassLoader extensionClassLoader =
IcebergArrowInputSourceReader.class.getClassLoader();
+ final ClassLoader originalClassLoader =
Thread.currentThread().getContextClassLoader();
+ try {
+ Thread.currentThread().setContextClassLoader(extensionClassLoader);
+ 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(),
+ extensionClassLoader
+ );
+ }
+ finally {
+ Thread.currentThread().setContextClassLoader(originalClassLoader);
+ }
+ }
+
+ 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);
Review Comment:
[P1] Preserve IGNORE residual semantics
`ArrowReader.open` hard-codes `.filter(task.residual())` for every
`FileScanTask`, and this scan never calls `ignoreResiduals()`. Consequently the
default `residualFilterMode=IGNORE` still drops nonmatching rows before Druid
sees them. The new `testWithEqualsFilter` expects all three rows for an
unpartitioned `name = alice` filter, but the Arrow iterator returns only the
two matches, so the behavior contradicts the documented standard mode and
prevents a Druid `transformSpec` from handling residual rows. Pass an
always-true residual/`ignoreResiduals()` for IGNORE while retaining FAIL
validation, or change the mode contract and tests.
--
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]
