FrankChen021 commented on code in PR #19510:
URL: https://github.com/apache/druid/pull/19510#discussion_r3990470949
##########
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:
AI is too strict on API changes, and I don't think this is true
--
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]
