WIP generic reader

.palantir/revapi.yml

@@ -1171,6 +1171,15 @@ acceptedBreaks:
         \ java.util.function.Function<T, java.lang.Long>, org.apache.iceberg.io.CloseableIterable<java.lang.Long>,\
         \ java.util.function.Consumer<T>)"
       justification: "Removing deprecated code"
+    org.apache.iceberg:iceberg-data:
+    - code: "java.class.removed"
+      old: "class org.apache.iceberg.data.BaseDeleteLoader"
+      justification: "Moved from iceberg-data to iceberg-core, so this should not\
+        \ cause issues"
+    - code: "java.class.removed"
+      old: "class org.apache.iceberg.data.InternalRecordWrapper"
+      justification: "Moved from iceberg-data to iceberg-core, so this should not\
+        \ cause issues"
   apache-iceberg-0.14.0:
     org.apache.iceberg:iceberg-api:
     - code: "java.class.defaultSerializationChanged"

arrow/src/main/java/org/apache/iceberg/arrow/vectorized/ArrowReader.java

@@ -33,16 +33,22 @@
 import org.apache.arrow.vector.VectorSchemaRoot;
 import org.apache.arrow.vector.types.Types.MinorType;
 import org.apache.iceberg.CombinedScanTask;
+import org.apache.iceberg.ContentScanTask;
+import org.apache.iceberg.DataFileReaderService;
+import org.apache.iceberg.DataFileReaderServiceRegistry;
 import org.apache.iceberg.FileFormat;
 import org.apache.iceberg.FileScanTask;
 import org.apache.iceberg.Schema;
+import org.apache.iceberg.Table;
 import org.apache.iceberg.TableScan;
+import org.apache.iceberg.data.DeleteFilter;
 import org.apache.iceberg.encryption.EncryptedFiles;
 import org.apache.iceberg.encryption.EncryptedInputFile;
 import org.apache.iceberg.encryption.EncryptionManager;
 import org.apache.iceberg.io.CloseableGroup;
 import org.apache.iceberg.io.CloseableIterable;
 import org.apache.iceberg.io.CloseableIterator;
+import org.apache.iceberg.io.FileFormatReadBuilder;
 import org.apache.iceberg.io.FileIO;
 import org.apache.iceberg.io.InputFile;
 import org.apache.iceberg.mapping.NameMappingParser;
@@ -322,16 +328,10 @@ CloseableIterator<ColumnarBatch> open(FileScanTask task) {
       InputFile location = getInputFile(task);
       Preconditions.checkNotNull(location, "Could not find InputFile associated with FileScanTask");
       if (task.file().format() == FileFormat.PARQUET) {
-        Parquet.ReadBuilder builder =
-            Parquet.read(location)
-                .project(expectedSchema)
+        FileFormatReadBuilder<?> builder =
+            DataFileReaderServiceRegistry.read(
+                    FileFormat.PARQUET, ColumnarBatch.class, location, expectedSchema)
                 .split(task.start(), task.length())
-                .createBatchedReaderFunc(
-                    fileSchema ->
-                        buildReader(
-                            expectedSchema,
-                            fileSchema, /* setArrowValidityVector */
-                            NullCheckingForGet.NULL_CHECKING_ENABLED))
                 .recordsPerBatch(batchSize)
                 .filter(task.residual())
                 .caseSensitive(caseSensitive);
@@ -388,4 +388,33 @@ private static ArrowBatchReader buildReader(
                   ArrowBatchReader::new));
     }
   }
+
+  public static class ReaderService implements DataFileReaderService {
+    @Override
+    public FileFormat format() {
+      return FileFormat.PARQUET;
+    }
+
+    @Override
+    public Class<?> returnType() {
+      return ColumnarBatch.class;
+    }
+
+    @Override
+    public FileFormatReadBuilder<?> builder(
+        InputFile inputFile,
+        ContentScanTask<?> task,
+        Schema readSchema,
+        Table table,
+        DeleteFilter<?> deleteFilter) {
+      return Parquet.read(inputFile)
+          .project(readSchema)
+          .createBatchedReaderFunc(
+              fileSchema ->
+                  VectorizedCombinedScanIterator.buildReader(
+                      readSchema,
+                      fileSchema, /* setArrowValidityVector */
+                      NullCheckingForGet.NULL_CHECKING_ENABLED));
+    }
+  }
 }

arrow/src/main/resources/META-INF/services/org.apache.iceberg.DataFileReaderService

@@ -0,0 +1,16 @@
+# 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.
+
+org.apache.iceberg.arrow.vectorized.ArrowReader$ReaderService

arrow/src/test/resources/META-INF/services/org.apache.iceberg.DataFileReaderService

@@ -0,0 +1,16 @@
+# 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.
+
+org.apache.iceberg.arrow.vectorized.ArrowReader$ReaderService

core/src/main/java/org/apache/iceberg/DataFileReaderService.java

@@ -0,0 +1,62 @@
+/*
+ * 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.iceberg;
+
+import org.apache.iceberg.data.DeleteFilter;
+import org.apache.iceberg.io.FileFormatReadBuilder;
+import org.apache.iceberg.io.InputFile;
+
+/**
+ * Service building readers. Implementations should be registered through the {@link
+ * java.util.ServiceLoader}. {@link DataFileReaderServiceRegistry} is used to collect and serve the
+ * reader implementations.
+ */
+public interface DataFileReaderService {
+  /**
+   * Returns the file format which is read by the readers.
+   *
+   * @return the input format of the reader
+   */
+  FileFormat format();
+
+  /**
+   * Returns the return type which is generated by the readers.
+   *
+   * @return the return type of the reader
+   */
+  Class<?> returnType();
+
+  /**
+   * Provides a reader for the given input file which returns objects with a given returnType.
+   *
+   * @param inputFile to read
+   * @param task to provide the values for metadata columns (_file_path, _spec_id, _partition)
+   * @param readSchema to use when reading the data file
+   * @param table to provide old partition specifications. Used for calculating values for
+   *     _partition column after specification changes
+   * @param deleteFilter is used when the delete record filtering is pushed down to the reader
+   * @return {@link FileFormatReadBuilder} for building the actual reader
+   */
+  FileFormatReadBuilder<?> builder(
+      InputFile inputFile,
+      ContentScanTask<?> task,
+      Schema readSchema,
+      Table table,
+      DeleteFilter<?> deleteFilter);
+}

core/src/main/java/org/apache/iceberg/DataFileReaderServiceRegistry.java

@@ -0,0 +1,144 @@
+/*
+ * 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.iceberg;
+
+import java.util.Map;
+import java.util.ServiceLoader;
+import org.apache.iceberg.data.DeleteFilter;
+import org.apache.iceberg.io.FileFormatReadBuilder;
+import org.apache.iceberg.io.InputFile;
+import org.apache.iceberg.relocated.com.google.common.base.MoreObjects;
+import org.apache.iceberg.relocated.com.google.common.base.Objects;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Registry which maintains the available {@link DataFileReaderService} implementations. Based on
+ * the file format and the required return type the registry returns the correct {@link
+ * FileFormatReadBuilder} implementations which could be used to generate the readers.
+ */
+public class DataFileReaderServiceRegistry {
+  private static final Logger LOG = LoggerFactory.getLogger(DataFileReaderServiceRegistry.class);
+  private static final Map<Key, DataFileReaderService> READ_BUILDERS = Maps.newConcurrentMap();
+
+  static {
+    ServiceLoader<DataFileReaderService> loader = ServiceLoader.load(DataFileReaderService.class);
+    for (DataFileReaderService service : loader) {
+      Key key = new Key(service.format(), service.returnType());
+      if (READ_BUILDERS.containsKey(key)) {
+        throw new IllegalArgumentException(
+            String.format(
+                "Service %s clashes with %s. Both serves %s",
+                service.getClass(), READ_BUILDERS.get(key), key));
+      }
+
+      READ_BUILDERS.putIfAbsent(key, service);
+    }
+
+    LOG.info("DataFileReaderServices found: {}", READ_BUILDERS);
+  }
+
+  private DataFileReaderServiceRegistry() {}
+
+  /**
+   * Provides a reader for the given {@link InputFile} which returns objects with a given
+   * returnType.
+   */
+  public static FileFormatReadBuilder<?> read(
+      FileFormat format, Class<?> returnType, InputFile inputFile, Schema readSchema) {
+    return read(format, returnType, inputFile, null, readSchema, null, null);
+  }
+
+  /**
+   * Provides a reader for the given {@link ContentScanTask} which returns objects with a given
+   * returnType.
+   */
+  public static FileFormatReadBuilder<?> read(
+      FileFormat format,
+      Class<?> returnType,
+      InputFile inputFile,
+      ContentScanTask<?> task,
+      Schema readSchema) {
+    return read(format, returnType, inputFile, task, readSchema, null, null);
+  }
+
+  /**
+   * Provides a reader for the given input file which returns objects with a given returnType.
+   *
+   * @param format of the file to read
+   * @param returnType returned by the reader
+   * @param inputFile to read
+   * @param task to provide the values for metadata columns (_file_path, _spec_id, _partition)
+   * @param readSchema to use when reading the data file
+   * @param table to provide old partition specifications. Used for calculating values for
+   *     _partition column after specification changes
+   * @param deleteFilter is used when the delete record filtering is pushed down to the reader
+   * @return {@link FileFormatReadBuilder} for building the actual reader
+   */
+  public static FileFormatReadBuilder<?> read(
+      FileFormat format,
+      Class<?> returnType,
+      InputFile inputFile,
+      ContentScanTask<?> task,
+      Schema readSchema,
+      Table table,
+      DeleteFilter<?> deleteFilter) {
+    return READ_BUILDERS
+        .get(new Key(format, returnType))
+        .builder(inputFile, task, readSchema, table, deleteFilter);
+  }
+
+  private static class Key {
+    private final FileFormat format;
+    private final Class<?> returnType;
+
+    private Key(FileFormat format, Class<?> returnType) {
+      this.format = format;
+      this.returnType = returnType;
+    }
+
+    @Override
+    public String toString() {
+      return MoreObjects.toStringHelper(this)
+          .add("format", format)
+          .add("returnType", returnType)
+          .toString();
+    }
+
+    @Override
+    public boolean equals(Object o) {
+      if (this == o) {
+        return true;
+      }
+
+      if (!(o instanceof Key)) {
+        return false;
+      }
+
+      Key other = (Key) o;
+      return Objects.equal(other.format, format) && Objects.equal(other.returnType, returnType);
+    }
+
+    @Override
+    public int hashCode() {
+      return Objects.hashCode(format, returnType);
+    }
+  }
+}