stevenzwu commented on code in PR #12298:
URL: https://github.com/apache/iceberg/pull/12298#discussion_r2379759933
##########
data/src/main/java/org/apache/iceberg/data/GenericAppenderFactory.java:
##########
@@ -71,6 +73,26 @@ public GenericAppenderFactory(
this(null, schema, spec, null, equalityFieldIds, eqDeleteRowSchema,
posDeleteRowSchema);
}
+ /**
+ * Constructor for GenericAppenderFactory.
+ *
+ * @param table iceberg table
+ * @param schema the schema of the records to write
+ * @param spec the partition spec of the records
+ * @param config the configuration for the writer
+ * @param equalityFieldIds the field ids for equality delete
+ * @param eqDeleteRowSchema the schema for equality delete rows
+ */
+ public GenericAppenderFactory(
+ Table table,
+ Schema schema,
+ PartitionSpec spec,
+ Map<String, String> config,
+ int[] equalityFieldIds,
+ Schema eqDeleteRowSchema) {
+ this(table, schema, spec, config, equalityFieldIds, eqDeleteRowSchema,
null);
Review Comment:
this new constructor still calls a deprecated constructor?
##########
data/src/main/java/org/apache/iceberg/data/GenericFileWriterFactory.java:
##########
@@ -50,62 +65,166 @@ class GenericFileWriterFactory extends
BaseFileWriterFactory<Record> {
super(
table,
dataFileFormat,
+ Record.class,
dataSchema,
dataSortOrder,
deleteFileFormat,
equalityFieldIds,
equalityDeleteRowSchema,
equalityDeleteSortOrder,
- positionDeleteRowSchema);
+ ImmutableMap.of(),
+ dataSchema,
+ equalityDeleteRowSchema);
+ this.table = table;
+ this.format = dataFileFormat;
+ this.positionDeleteRowSchema = positionDeleteRowSchema;
}
static Builder builderFor(Table table) {
return new Builder(table);
}
- @Override
+ /**
+ * @deprecated Since 1.10.0, will be removed in 1.11.0. It won't be called
starting 1.10.0 as the
+ * configuration is done by the {@link FormatModelRegistry}.
+ */
+ @Deprecated
protected void configureDataWrite(Avro.DataWriteBuilder builder) {
- builder.createWriterFunc(DataWriter::create);
+ throwUnsupportedOperationException();
}
- @Override
+ /**
+ * @deprecated Since 1.10.0, will be removed in 1.11.0. It won't be called
starting 1.10.0 as the
+ * configuration is done by the {@link FormatModelRegistry}.
+ */
+ @Deprecated
protected void configureEqualityDelete(Avro.DeleteWriteBuilder builder) {
- builder.createWriterFunc(DataWriter::create);
+ throwUnsupportedOperationException();
}
- @Override
+ /**
+ * @deprecated Since 1.10.0, will be removed in 1.11.0. It won't be called
starting 1.10.0 as the
+ * configuration is done by the {@link FormatModelRegistry}.
+ */
+ @Deprecated
protected void configurePositionDelete(Avro.DeleteWriteBuilder builder) {
- builder.createWriterFunc(DataWriter::create);
+ throwUnsupportedOperationException();
}
- @Override
+ /**
+ * @deprecated Since 1.10.0, will be removed in 1.11.0. It won't be called
starting 1.10.0 as the
+ * configuration is done by the {@link FormatModelRegistry}.
+ */
+ @Deprecated
protected void configureDataWrite(Parquet.DataWriteBuilder builder) {
- builder.createWriterFunc(GenericParquetWriter::create);
+ throwUnsupportedOperationException();
}
- @Override
+ /**
+ * @deprecated Since 1.10.0, will be removed in 1.11.0. It won't be called
starting 1.10.0 as the
+ * configuration is done by the {@link FormatModelRegistry}.
+ */
+ @Deprecated
protected void configureEqualityDelete(Parquet.DeleteWriteBuilder builder) {
- builder.createWriterFunc(GenericParquetWriter::create);
+ throwUnsupportedOperationException();
}
- @Override
+ /**
+ * @deprecated Since 1.10.0, will be removed in 1.11.0. It won't be called
starting 1.10.0 as the
+ * configuration is done by the {@link FormatModelRegistry}.
+ */
+ @Deprecated
protected void configurePositionDelete(Parquet.DeleteWriteBuilder builder) {
- builder.createWriterFunc(GenericParquetWriter::create);
+ throwUnsupportedOperationException();
}
- @Override
+ /**
+ * @deprecated Since 1.10.0, will be removed in 1.11.0. It won't be called
starting 1.10.0 as the
+ * configuration is done by the {@link FormatModelRegistry}.
+ */
+ @Deprecated
protected void configureDataWrite(ORC.DataWriteBuilder builder) {
- builder.createWriterFunc(GenericOrcWriter::buildWriter);
+ throwUnsupportedOperationException();
}
- @Override
+ /**
+ * @deprecated Since 1.10.0, will be removed in 1.11.0. It won't be called
starting 1.10.0 as the
+ * configuration is done by the {@link FormatModelRegistry}.
+ */
+ @Deprecated
protected void configureEqualityDelete(ORC.DeleteWriteBuilder builder) {
- builder.createWriterFunc(GenericOrcWriter::buildWriter);
+ throwUnsupportedOperationException();
}
- @Override
+ /**
+ * @deprecated Since 1.10.0, will be removed in 1.11.0. It won't be called
starting 1.10.0 as the
+ * configuration is done by the {@link FormatModelRegistry}.
+ */
+ @Deprecated
protected void configurePositionDelete(ORC.DeleteWriteBuilder builder) {
- builder.createWriterFunc(GenericOrcWriter::buildWriter);
+ throwUnsupportedOperationException();
+ }
+
+ private void throwUnsupportedOperationException() {
+ throw new UnsupportedOperationException(
+ "Method is deprecated and should not be called. "
+ + "Configuration is already done by the ObjectModelRegistry.");
+ }
+
+ @Override
+ public PositionDeleteWriter<Record> newPositionDeleteWriter(
+ EncryptedOutputFile file, PartitionSpec spec, StructLike partition) {
+ if (positionDeleteRowSchema == null) {
+ return super.newPositionDeleteWriter(file, spec, partition);
+ } else {
+ LOG.info(
Review Comment:
Spark integration just silently ignore the `positionDeleteRowSchema`
directly, because we thought nobody is using the embedded row. why do we need
to support the handling here? Hence wondering if the override is necessary here.
##########
parquet/src/main/java/org/apache/iceberg/parquet/Parquet.java:
##########
@@ -136,6 +138,15 @@ private Parquet() {}
"parquet.read.support.class",
"parquet.crypto.factory.class");
+ private static final int MAX_RECORDS_PER_BATCH_DEFAULT = 10000;
+
+ public static final String WRITER_VERSION_KEY = "parquet.writer.version";
+
+ /**
+ * @deprecated Since 1.10.0, will be removed in 1.11.0. Use {@link
+ * FormatModelRegistry#writeBuilder(FileFormat, Class,
EncryptedOutputFile)} instead.
+ */
+ @Deprecated
Review Comment:
I thought Ryan raised a question in some meeting if we should force
everything to the `FormatModelRegistry` route. can clients still be able to
call Parquet APIs directly?
##########
parquet/src/main/java/org/apache/iceberg/parquet/Parquet.java:
##########
@@ -260,21 +269,147 @@ public WriteBuilder overwrite() {
}
public WriteBuilder overwrite(boolean enabled) {
- this.writeMode = enabled ? ParquetFileWriter.Mode.OVERWRITE :
ParquetFileWriter.Mode.CREATE;
+ impl.writeMode = enabled ? ParquetFileWriter.Mode.OVERWRITE :
ParquetFileWriter.Mode.CREATE;
return this;
}
public WriteBuilder writerVersion(WriterVersion version) {
- this.writerVersion = version;
+ impl.set(WRITER_VERSION_KEY, version.name());
return this;
}
public WriteBuilder withFileEncryptionKey(ByteBuffer encryptionKey) {
- this.fileEncryptionKey = encryptionKey;
+ impl.fileEncryptionKey(encryptionKey);
return this;
}
public WriteBuilder withAADPrefix(ByteBuffer aadPrefix) {
+ impl.fileAADPrefix(aadPrefix);
+ return this;
+ }
+
+ /*
+ * Sets the writer version. Default value is PARQUET_1_0 (v1).
+ */
+ @VisibleForTesting
+ WriteBuilder withWriterVersion(WriterVersion version) {
Review Comment:
is this a new behavior we are introducing here?
##########
parquet/src/main/java/org/apache/iceberg/parquet/ReadConf.java:
##########
@@ -59,17 +59,16 @@ class ReadConf<T> {
// List of column chunk metadata for each row group
private final List<Map<ColumnPath, ColumnChunkMetaData>>
columnChunkMetaDataForRowGroups;
- @SuppressWarnings("unchecked")
ReadConf(
InputFile file,
ParquetReadOptions options,
Schema expectedSchema,
Expression filter,
- Function<MessageType, ParquetValueReader<?>> readerFunc,
- Function<MessageType, VectorizedReader<?>> batchedReaderFunc,
+ Function<MessageType, ParquetValueReader<T>> readerFunc,
+ Function<MessageType, VectorizedReader<T>> batchedReaderFunc,
Review Comment:
we don't need read conf for the new `BatchReaderFunction` interface?
##########
parquet/src/main/java/org/apache/iceberg/parquet/Parquet.java:
##########
@@ -1210,47 +1402,64 @@ public ReadBuilder filter(Expression newFilter) {
*/
@Deprecated
public ReadBuilder readSupport(ReadSupport<?> newFilterSupport) {
- this.readSupport = newFilterSupport;
+ impl.readSupport = newFilterSupport;
return this;
}
public ReadBuilder createReaderFunc(
Function<MessageType, ParquetValueReader<?>> newReaderFunction) {
- Preconditions.checkArgument(
- this.batchedReaderFunc == null,
- "Cannot set reader function: batched reader function already set");
- Preconditions.checkArgument(
- this.readerFuncWithSchema == null,
- "Cannot set reader function: 2-argument reader function already
set");
- this.readerFunc = newReaderFunction;
+ Preconditions.checkState(
+ impl.readerFuncWithSchema == null
+ && impl.readerFunction == null
+ && impl.batchedReaderFunc == null
+ && impl.batchReaderFunction == null,
+ "Cannot set multiple read builder functions");
+ if (newReaderFunction != null) {
+ impl.readerFunc = m -> (ParquetValueReader<Object>)
newReaderFunction.apply(m);
Review Comment:
nit: `m` -> `messageType`
##########
parquet/src/main/java/org/apache/iceberg/parquet/Parquet.java:
##########
@@ -1315,61 +1676,12 @@ public <D> CloseableIterable<D> build() {
Preconditions.checkState(fileAADPrefix == null, "AAD prefix set with
null encryption key");
}
- if (readerFunc != null || readerFuncWithSchema != null ||
batchedReaderFunc != null) {
- ParquetReadOptions.Builder optionsBuilder;
- if (file instanceof HadoopInputFile) {
- // remove read properties already set that may conflict with this
read
- Configuration conf = new Configuration(((HadoopInputFile)
file).getConf());
- for (String property : READ_PROPERTIES_TO_REMOVE) {
- conf.unset(property);
- }
- optionsBuilder = HadoopReadOptions.builder(conf);
- } else {
- optionsBuilder = ParquetReadOptions.builder(new
PlainParquetConfiguration());
- }
-
- for (Map.Entry<String, String> entry : properties.entrySet()) {
- optionsBuilder.set(entry.getKey(), entry.getValue());
- }
-
- if (start != null) {
- optionsBuilder.withRange(start, start + length);
- }
-
- if (fileDecryptionProperties != null) {
- optionsBuilder.withDecryption(fileDecryptionProperties);
- }
-
- ParquetReadOptions options = optionsBuilder.build();
-
- NameMapping mapping;
- if (nameMapping != null) {
- mapping = nameMapping;
- } else if
(SystemConfigs.NETFLIX_UNSAFE_PARQUET_ID_FALLBACK_ENABLED.value()) {
- mapping = null;
- } else {
- mapping = NameMapping.empty();
- }
-
- if (batchedReaderFunc != null) {
- return new VectorizedParquetReader<>(
- file,
- schema,
- options,
- batchedReaderFunc,
- mapping,
- filter,
- reuseContainers,
- caseSensitive,
- maxRecordsPerBatch);
- } else {
- Function<MessageType, ParquetValueReader<?>> readBuilder =
- readerFuncWithSchema != null
- ? fileType -> readerFuncWithSchema.apply(schema, fileType)
- : readerFunc;
- return new org.apache.iceberg.parquet.ParquetReader<>(
- file, schema, options, readBuilder, mapping, filter,
reuseContainers, caseSensitive);
- }
+ if (readerFunc != null
+ || readerFuncWithSchema != null
+ || batchedReaderFunc != null
+ || readerFunction != null
+ || batchReaderFunction != null) {
+ return buildFunctionBasedReader(options(fileDecryptionProperties));
Review Comment:
nit: `buildFunctionBasedReader` -> `buildFromReaderFunction` to match the
return value `CloseableIterable<D>` a bit better. technically, it is not a
reader. it is the read output.
##########
data/src/main/java/org/apache/iceberg/data/GenericFileWriterFactory.java:
##########
@@ -50,62 +65,166 @@ class GenericFileWriterFactory extends
BaseFileWriterFactory<Record> {
super(
table,
dataFileFormat,
+ Record.class,
dataSchema,
dataSortOrder,
deleteFileFormat,
equalityFieldIds,
equalityDeleteRowSchema,
equalityDeleteSortOrder,
- positionDeleteRowSchema);
+ ImmutableMap.of(),
+ dataSchema,
+ equalityDeleteRowSchema);
+ this.table = table;
+ this.format = dataFileFormat;
+ this.positionDeleteRowSchema = positionDeleteRowSchema;
}
static Builder builderFor(Table table) {
return new Builder(table);
}
- @Override
+ /**
+ * @deprecated Since 1.10.0, will be removed in 1.11.0. It won't be called
starting 1.10.0 as the
+ * configuration is done by the {@link FormatModelRegistry}.
+ */
+ @Deprecated
protected void configureDataWrite(Avro.DataWriteBuilder builder) {
- builder.createWriterFunc(DataWriter::create);
+ throwUnsupportedOperationException();
}
- @Override
+ /**
+ * @deprecated Since 1.10.0, will be removed in 1.11.0. It won't be called
starting 1.10.0 as the
+ * configuration is done by the {@link FormatModelRegistry}.
+ */
+ @Deprecated
protected void configureEqualityDelete(Avro.DeleteWriteBuilder builder) {
- builder.createWriterFunc(DataWriter::create);
+ throwUnsupportedOperationException();
}
- @Override
+ /**
+ * @deprecated Since 1.10.0, will be removed in 1.11.0. It won't be called
starting 1.10.0 as the
+ * configuration is done by the {@link FormatModelRegistry}.
+ */
+ @Deprecated
protected void configurePositionDelete(Avro.DeleteWriteBuilder builder) {
- builder.createWriterFunc(DataWriter::create);
+ throwUnsupportedOperationException();
}
- @Override
+ /**
+ * @deprecated Since 1.10.0, will be removed in 1.11.0. It won't be called
starting 1.10.0 as the
+ * configuration is done by the {@link FormatModelRegistry}.
+ */
+ @Deprecated
protected void configureDataWrite(Parquet.DataWriteBuilder builder) {
- builder.createWriterFunc(GenericParquetWriter::create);
+ throwUnsupportedOperationException();
}
- @Override
+ /**
+ * @deprecated Since 1.10.0, will be removed in 1.11.0. It won't be called
starting 1.10.0 as the
+ * configuration is done by the {@link FormatModelRegistry}.
+ */
+ @Deprecated
protected void configureEqualityDelete(Parquet.DeleteWriteBuilder builder) {
- builder.createWriterFunc(GenericParquetWriter::create);
+ throwUnsupportedOperationException();
}
- @Override
+ /**
+ * @deprecated Since 1.10.0, will be removed in 1.11.0. It won't be called
starting 1.10.0 as the
+ * configuration is done by the {@link FormatModelRegistry}.
+ */
+ @Deprecated
protected void configurePositionDelete(Parquet.DeleteWriteBuilder builder) {
- builder.createWriterFunc(GenericParquetWriter::create);
+ throwUnsupportedOperationException();
}
- @Override
+ /**
+ * @deprecated Since 1.10.0, will be removed in 1.11.0. It won't be called
starting 1.10.0 as the
+ * configuration is done by the {@link FormatModelRegistry}.
+ */
+ @Deprecated
protected void configureDataWrite(ORC.DataWriteBuilder builder) {
- builder.createWriterFunc(GenericOrcWriter::buildWriter);
+ throwUnsupportedOperationException();
}
- @Override
+ /**
+ * @deprecated Since 1.10.0, will be removed in 1.11.0. It won't be called
starting 1.10.0 as the
+ * configuration is done by the {@link FormatModelRegistry}.
+ */
+ @Deprecated
protected void configureEqualityDelete(ORC.DeleteWriteBuilder builder) {
- builder.createWriterFunc(GenericOrcWriter::buildWriter);
+ throwUnsupportedOperationException();
}
- @Override
+ /**
+ * @deprecated Since 1.10.0, will be removed in 1.11.0. It won't be called
starting 1.10.0 as the
+ * configuration is done by the {@link FormatModelRegistry}.
+ */
+ @Deprecated
protected void configurePositionDelete(ORC.DeleteWriteBuilder builder) {
- builder.createWriterFunc(GenericOrcWriter::buildWriter);
+ throwUnsupportedOperationException();
+ }
+
+ private void throwUnsupportedOperationException() {
+ throw new UnsupportedOperationException(
+ "Method is deprecated and should not be called. "
+ + "Configuration is already done by the ObjectModelRegistry.");
Review Comment:
ObjectModelRegistry -> format model registry
we usually don't use class name in error msg
##########
data/src/test/java/org/apache/iceberg/io/TestAppenderFactory.java:
##########
@@ -259,6 +260,7 @@ public void testPosDeleteWriter() throws IOException {
.isEqualTo(expectedRowSet(expected));
}
+ @Disabled("Deprecated API")
Review Comment:
can we separate out the embedded row deprecation to a separate PR? we can
probably merge that one in fairly quickly.
##########
data/src/main/java/org/apache/iceberg/data/GenericFormatModels.java:
##########
@@ -0,0 +1,94 @@
+/*
+ * 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.data;
+
+import org.apache.iceberg.avro.AvroFormatModel;
+import org.apache.iceberg.data.avro.DataWriter;
+import org.apache.iceberg.data.avro.PlannedDataReader;
+import org.apache.iceberg.data.orc.GenericOrcReader;
+import org.apache.iceberg.data.orc.GenericOrcWriter;
+import org.apache.iceberg.data.parquet.GenericParquetReaders;
+import org.apache.iceberg.data.parquet.GenericParquetWriter;
+import org.apache.iceberg.deletes.PositionDelete;
+import org.apache.iceberg.orc.ORCFormatModel;
+import org.apache.iceberg.parquet.ParquetFormatModel;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class GenericFormatModels {
+ private static final Logger LOG =
LoggerFactory.getLogger(GenericFormatModels.class);
+
+ public static void register() {
+ // ORC, Parquet are optional dependencies. If they are not present, we
should just log and
+ // ignore NoClassDefFoundErrors
+ registerAvro();
+ registerParquet();
+ registerOrc();
+ }
+
+ private static void registerParquet() {
+ logAngIgnoreNoClassDefFoundError(
+ () ->
+ FormatModelRegistry.register(
+ new ParquetFormatModel<>(
+ Record.class,
+ GenericParquetReaders::buildReader,
+ (schema, messageType, inputType) ->
+ GenericParquetWriter.create(schema, messageType))));
+ logAngIgnoreNoClassDefFoundError(
+ () -> FormatModelRegistry.register(new
ParquetFormatModel<>(PositionDelete.class)));
+ }
+
+ private static void registerAvro() {
+ logAngIgnoreNoClassDefFoundError(
+ () ->
+ FormatModelRegistry.register(
+ new AvroFormatModel<>(
+ Record.class,
+ PlannedDataReader::create,
+ (schema, inputSchema) -> DataWriter.create(schema))));
+ logAngIgnoreNoClassDefFoundError(
+ () -> FormatModelRegistry.register(new
AvroFormatModel<>(PositionDelete.class)));
+ }
+
+ private static void registerOrc() {
+ logAngIgnoreNoClassDefFoundError(
+ () ->
+ FormatModelRegistry.register(
+ new ORCFormatModel<>(
+ Record.class,
+ GenericOrcReader::buildReader,
+ (schema, typeDescription, unused) ->
+ GenericOrcWriter.buildWriter(schema,
typeDescription))));
+ logAngIgnoreNoClassDefFoundError(
+ () -> FormatModelRegistry.register(new
ORCFormatModel<>(PositionDelete.class)));
+ }
+
+ private GenericFormatModels() {}
+
+ @SuppressWarnings("CatchBlockLogException")
+ private static void logAngIgnoreNoClassDefFoundError(Runnable runnable) {
+ try {
+ runnable.run();
+ } catch (NoClassDefFoundError e) {
+ // Log the exception and ignore it
+ LOG.info("Exception occurred when trying to register object models: {}",
e.getMessage());
Review Comment:
nit: object models -> format model
##########
parquet/src/main/java/org/apache/iceberg/parquet/Parquet.java:
##########
@@ -1210,47 +1402,64 @@ public ReadBuilder filter(Expression newFilter) {
*/
@Deprecated
public ReadBuilder readSupport(ReadSupport<?> newFilterSupport) {
- this.readSupport = newFilterSupport;
+ impl.readSupport = newFilterSupport;
return this;
}
public ReadBuilder createReaderFunc(
Function<MessageType, ParquetValueReader<?>> newReaderFunction) {
- Preconditions.checkArgument(
- this.batchedReaderFunc == null,
- "Cannot set reader function: batched reader function already set");
- Preconditions.checkArgument(
- this.readerFuncWithSchema == null,
- "Cannot set reader function: 2-argument reader function already
set");
- this.readerFunc = newReaderFunction;
+ Preconditions.checkState(
+ impl.readerFuncWithSchema == null
+ && impl.readerFunction == null
+ && impl.batchedReaderFunc == null
+ && impl.batchReaderFunction == null,
+ "Cannot set multiple read builder functions");
+ if (newReaderFunction != null) {
+ impl.readerFunc = m -> (ParquetValueReader<Object>)
newReaderFunction.apply(m);
+ } else {
+ impl.readerFunc = null;
+ }
+
return this;
}
public ReadBuilder createReaderFunc(
BiFunction<Schema, MessageType, ParquetValueReader<?>>
newReaderFunction) {
- Preconditions.checkArgument(
- this.readerFunc == null,
- "Cannot set 2-argument reader function: reader function already
set");
- Preconditions.checkArgument(
- this.batchedReaderFunc == null,
- "Cannot set 2-argument reader function: batched reader function
already set");
- this.readerFuncWithSchema = newReaderFunction;
+ Preconditions.checkState(
+ impl.readerFunc == null
+ && impl.readerFunction == null
+ && impl.batchedReaderFunc == null
Review Comment:
why do we have both batchedReaderFunc and batchReaderFunction?
##########
parquet/src/main/java/org/apache/iceberg/parquet/ParquetFormatModel.java:
##########
@@ -0,0 +1,105 @@
+/*
+ * 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.parquet;
+
+import java.util.Map;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.io.FormatModel;
+import org.apache.iceberg.io.InputFile;
+import org.apache.iceberg.io.OutputFile;
+import org.apache.parquet.schema.MessageType;
+
+public class ParquetFormatModel<D, S, F> implements FormatModel<D, S> {
+ private final Class<D> type;
+ private final ReaderFunction<D> readerFunction;
+ private final BatchReaderFunction<D, F> batchReaderFunction;
+ private final WriterFunction<D, S> writerFunction;
+
+ private ParquetFormatModel(
+ Class<D> type,
+ ReaderFunction<D> readerFunction,
+ BatchReaderFunction<D, F> batchReaderFunction,
+ WriterFunction<D, S> writerFunction) {
+ this.type = type;
+ this.readerFunction = readerFunction;
+ this.batchReaderFunction = batchReaderFunction;
+ this.writerFunction = writerFunction;
+ }
+
+ public ParquetFormatModel(Class<D> type) {
+ this(type, null, null, null);
+ }
+
+ public ParquetFormatModel(
+ Class<D> type, ReaderFunction<D> readerFunction, WriterFunction<D, S>
writerFunction) {
+ this(type, readerFunction, null, writerFunction);
+ }
+
+ public ParquetFormatModel(Class<D> type, BatchReaderFunction<D, F>
batchReaderFunction) {
+ this(type, null, batchReaderFunction, null);
+ }
+
+ @Override
+ public FileFormat format() {
+ return FileFormat.PARQUET;
+ }
+
+ @Override
+ public Class<D> type() {
+ return type;
+ }
+
+ @Override
+ public org.apache.iceberg.io.WriteBuilder<D, S> writeBuilder(OutputFile
outputFile) {
+ return new Parquet.WriteBuilderImpl<D,
S>(outputFile).writerFunction(writerFunction);
+ }
+
+ @Override
+ public org.apache.iceberg.io.ReadBuilder<D, S> readBuilder(InputFile
inputFile) {
+ if (batchReaderFunction != null) {
+ return new Parquet.ReadBuilderImpl<D, S, F>(inputFile)
+ .batchReaderFunction(batchReaderFunction);
+ } else {
+ return new Parquet.ReadBuilderImpl<D, S,
F>(inputFile).readerFunction(readerFunction);
+ }
+ }
+
+ public interface ReaderFunction<D> {
Review Comment:
nit: annotate these interfaces with `@FunctionalInterface`
--
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]
