This is an automated email from the ASF dual-hosted git repository. lzljs3620320 pushed a commit to branch release-0.2 in repository https://gitbox.apache.org/repos/asf/flink-table-store.git
commit 1973b737fc97fe1ecc83ee2c80f3e9253557ed7a Author: shammon <[email protected]> AuthorDate: Wed Sep 21 11:37:28 2022 +0800 [FLINK-29345] Create reusing reader/writer config in orc format This closes #296 --- .../store/format/orc/OrcBulkWriterFactory.java | 116 +++++++++++++++++++++ .../table/store/format/orc/OrcFileFormat.java | 35 ++++--- .../store/format/orc/OrcBulkWriterFactoryTest.java | 83 +++++++++++++++ .../table/store/format/orc/OrcFileFormatTest.java | 9 +- 4 files changed, 225 insertions(+), 18 deletions(-) diff --git a/flink-table-store-format/src/main/java/org/apache/flink/table/store/format/orc/OrcBulkWriterFactory.java b/flink-table-store-format/src/main/java/org/apache/flink/table/store/format/orc/OrcBulkWriterFactory.java new file mode 100644 index 00000000..b6670392 --- /dev/null +++ b/flink-table-store-format/src/main/java/org/apache/flink/table/store/format/orc/OrcBulkWriterFactory.java @@ -0,0 +1,116 @@ +/* + * 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.flink.table.store.format.orc; + +import org.apache.flink.annotation.VisibleForTesting; +import org.apache.flink.api.common.serialization.BulkWriter; +import org.apache.flink.core.fs.FSDataOutputStream; +import org.apache.flink.orc.vector.Vectorizer; +import org.apache.flink.orc.writer.PhysicalWriterImpl; + +import org.apache.hadoop.fs.Path; +import org.apache.hadoop.hive.ql.exec.vector.VectorizedRowBatch; +import org.apache.orc.OrcFile; +import org.apache.orc.Writer; +import org.apache.orc.impl.WriterImpl; + +import java.io.IOException; +import java.util.UUID; + +import static org.apache.flink.util.Preconditions.checkNotNull; + +/** + * Orc {@link BulkWriter.Factory}. The main code is copied from Flink {@code OrcBulkWriterFactory}. + */ +public class OrcBulkWriterFactory<T> implements BulkWriter.Factory<T> { + + private final Vectorizer<T> vectorizer; + private final OrcFile.WriterOptions writerOptions; + + /** + * Creates a new OrcBulkWriterFactory using the provided Vectorizer, ORC WriterOptions. + * + * @param vectorizer The vectorizer implementation to convert input record to a + * VectorizerRowBatch. + * @param writerOptions ORC WriterOptions. + */ + public OrcBulkWriterFactory(Vectorizer<T> vectorizer, OrcFile.WriterOptions writerOptions) { + this.vectorizer = checkNotNull(vectorizer); + this.writerOptions = checkNotNull(writerOptions); + } + + @Override + public BulkWriter<T> create(FSDataOutputStream out) throws IOException { + OrcFile.WriterOptions opts = getWriterOptions(); + opts.physicalWriter(new PhysicalWriterImpl(out, opts)); + + // The path of the Writer is not used to indicate the destination file + // in this case since we have used a dedicated physical writer to write + // to the give output stream directly. However, the path would be used as + // the key of writer in the ORC memory manager, thus we need to make it unique. + Path unusedPath = new Path(UUID.randomUUID().toString()); + return new OrcBulkWriter<>(vectorizer, new WriterImpl(null, unusedPath, opts)); + } + + @VisibleForTesting + protected OrcFile.WriterOptions getWriterOptions() { + return writerOptions; + } + + /** Orc {@link BulkWriter}. The main code is copied from Flink {@code OrcBulkWriter}. */ + private static class OrcBulkWriter<T> implements BulkWriter<T> { + + private final Writer writer; + private final Vectorizer<T> vectorizer; + private final VectorizedRowBatch rowBatch; + + OrcBulkWriter(Vectorizer<T> vectorizer, Writer writer) { + this.vectorizer = checkNotNull(vectorizer); + this.writer = checkNotNull(writer); + this.rowBatch = vectorizer.getSchema().createRowBatch(); + + // Configure the vectorizer with the writer so that users can add + // metadata on the fly through the Vectorizer#vectorize(...) method. + this.vectorizer.setWriter(this.writer); + } + + @Override + public void addElement(T element) throws IOException { + vectorizer.vectorize(element, rowBatch); + if (rowBatch.size == rowBatch.getMaxSize()) { + writer.addRowBatch(rowBatch); + rowBatch.reset(); + } + } + + @Override + public void flush() throws IOException { + if (rowBatch.size != 0) { + writer.addRowBatch(rowBatch); + rowBatch.reset(); + } + } + + @Override + public void finish() throws IOException { + flush(); + writer.close(); + } + } +} diff --git a/flink-table-store-format/src/main/java/org/apache/flink/table/store/format/orc/OrcFileFormat.java b/flink-table-store-format/src/main/java/org/apache/flink/table/store/format/orc/OrcFileFormat.java index 880074c2..3192cf56 100644 --- a/flink-table-store-format/src/main/java/org/apache/flink/table/store/format/orc/OrcFileFormat.java +++ b/flink-table-store-format/src/main/java/org/apache/flink/table/store/format/orc/OrcFileFormat.java @@ -27,7 +27,8 @@ import org.apache.flink.connector.file.src.reader.BulkFormat; import org.apache.flink.orc.OrcFilters; import org.apache.flink.orc.OrcSplitReaderUtil; import org.apache.flink.orc.vector.RowDataVectorizer; -import org.apache.flink.orc.writer.OrcBulkWriterFactory; +import org.apache.flink.orc.vector.Vectorizer; +import org.apache.flink.orc.writer.ThreadLocalClassLoaderConfiguration; import org.apache.flink.table.data.RowData; import org.apache.flink.table.store.file.predicate.Predicate; import org.apache.flink.table.store.format.FileFormat; @@ -36,6 +37,7 @@ import org.apache.flink.table.store.utils.Projection; import org.apache.flink.table.types.logical.LogicalType; import org.apache.flink.table.types.logical.RowType; +import org.apache.orc.OrcFile; import org.apache.orc.TypeDescription; import java.util.ArrayList; @@ -48,16 +50,24 @@ import static org.apache.flink.table.store.format.orc.OrcFileFormatFactory.IDENT /** Orc {@link FileFormat}. The main code is copied from Flink {@code OrcFileFormatFactory}. */ public class OrcFileFormat extends FileFormat { - private final Configuration formatOptions; + private final Properties orcProperties; + private final org.apache.hadoop.conf.Configuration readerConf; + private final org.apache.hadoop.conf.Configuration writerConf; public OrcFileFormat(Configuration formatOptions) { super(org.apache.flink.orc.OrcFileFormatFactory.IDENTIFIER); - this.formatOptions = formatOptions; + this.orcProperties = getOrcProperties(formatOptions); + this.readerConf = new org.apache.hadoop.conf.Configuration(); + this.orcProperties.forEach((k, v) -> readerConf.set(k.toString(), v.toString())); + + org.apache.hadoop.conf.Configuration conf = new org.apache.hadoop.conf.Configuration(); + this.writerConf = new ThreadLocalClassLoaderConfiguration(); + conf.forEach(entry -> writerConf.set(entry.getKey(), entry.getValue())); } @VisibleForTesting - Configuration formatOptions() { - return formatOptions; + Properties orcProperties() { + return orcProperties; } @Override @@ -79,12 +89,8 @@ public class OrcFileFormat extends FileFormat { } } - Properties properties = getOrcProperties(formatOptions); - org.apache.hadoop.conf.Configuration conf = new org.apache.hadoop.conf.Configuration(); - properties.forEach((k, v) -> conf.set(k.toString(), v.toString())); - return OrcInputFormatFactory.create( - conf, type, Projection.of(projection).toTopLevelIndexes(), orcPredicates); + readerConf, type, Projection.of(projection).toTopLevelIndexes(), orcPredicates); } @Override @@ -92,11 +98,12 @@ public class OrcFileFormat extends FileFormat { LogicalType[] orcTypes = type.getChildren().toArray(new LogicalType[0]); TypeDescription typeDescription = OrcSplitReaderUtil.logicalTypeToOrcType(type); + Vectorizer<RowData> vectorizer = + new RowDataVectorizer(typeDescription.toString(), orcTypes); + OrcFile.WriterOptions writerOptions = OrcFile.writerOptions(orcProperties, writerConf); + writerOptions.setSchema(vectorizer.getSchema()); - return new OrcBulkWriterFactory<>( - new RowDataVectorizer(typeDescription.toString(), orcTypes), - getOrcProperties(formatOptions), - new org.apache.hadoop.conf.Configuration()); + return new OrcBulkWriterFactory<>(vectorizer, writerOptions); } private static Properties getOrcProperties(ReadableConfig options) { diff --git a/flink-table-store-format/src/test/java/org/apache/flink/table/store/format/orc/OrcBulkWriterFactoryTest.java b/flink-table-store-format/src/test/java/org/apache/flink/table/store/format/orc/OrcBulkWriterFactoryTest.java new file mode 100644 index 00000000..374bc1c6 --- /dev/null +++ b/flink-table-store-format/src/test/java/org/apache/flink/table/store/format/orc/OrcBulkWriterFactoryTest.java @@ -0,0 +1,83 @@ +/* + * 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.flink.table.store.format.orc; + +import org.apache.flink.configuration.Configuration; +import org.apache.flink.core.fs.local.LocalDataOutputStream; +import org.apache.flink.table.data.RowData; +import org.apache.flink.table.types.logical.IntType; +import org.apache.flink.table.types.logical.RowType; +import org.apache.flink.table.types.logical.VarCharType; + +import org.apache.hadoop.fs.Path; +import org.apache.orc.MemoryManager; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +import java.io.File; +import java.io.IOException; +import java.util.ArrayList; +import java.util.List; +import java.util.UUID; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotEquals; + +/** Test for {@link OrcBulkWriterFactory}. */ +public class OrcBulkWriterFactoryTest { + @TempDir File tempDir; + + @Test + public void testNotOverrideInMemoryManager() throws IOException { + OrcFileFormat orcFileFormat = new OrcFileFormat(new Configuration()); + OrcBulkWriterFactory<RowData> factory = + (OrcBulkWriterFactory<RowData>) + orcFileFormat.createWriterFactory( + RowType.of(new VarCharType(), new IntType())); + + TestMemoryManager memoryManager = new TestMemoryManager(); + factory.getWriterOptions().memory(memoryManager); + + factory.create(new LocalDataOutputStream(new File(tempDir, UUID.randomUUID().toString()))); + factory.create(new LocalDataOutputStream(new File(tempDir, UUID.randomUUID().toString()))); + + List<Path> addedWriterPath = memoryManager.getAddedWriterPath(); + assertEquals(2, addedWriterPath.size()); + assertNotEquals(addedWriterPath.get(1), addedWriterPath.get(0)); + } + + private static class TestMemoryManager implements MemoryManager { + private final List<Path> addedWriterPath = new ArrayList<>(); + + @Override + public void addWriter(Path path, long requestedAllocation, Callback callback) { + addedWriterPath.add(path); + } + + public List<Path> getAddedWriterPath() { + return addedWriterPath; + } + + @Override + public void removeWriter(Path path) {} + + @Override + public void addedRow(int rows) {} + } +} diff --git a/flink-table-store-format/src/test/java/org/apache/flink/table/store/format/orc/OrcFileFormatTest.java b/flink-table-store-format/src/test/java/org/apache/flink/table/store/format/orc/OrcFileFormatTest.java index b99d337f..ba1dbacb 100644 --- a/flink-table-store-format/src/test/java/org/apache/flink/table/store/format/orc/OrcFileFormatTest.java +++ b/flink-table-store-format/src/test/java/org/apache/flink/table/store/format/orc/OrcFileFormatTest.java @@ -22,6 +22,7 @@ import org.apache.flink.configuration.Configuration; import org.junit.jupiter.api.Test; +import static org.apache.flink.table.store.format.orc.OrcFileFormatFactory.IDENTIFIER; import static org.assertj.core.api.Assertions.assertThat; /** Test for {@link OrcFileFormatFactory}. */ @@ -32,8 +33,8 @@ public class OrcFileFormatTest { Configuration options = new Configuration(); options.setString("haha", "1"); OrcFileFormat orc = new OrcFileFormatFactory().create(options); - assertThat(orc.formatOptions().getString("haha", "")).isEqualTo("1"); - assertThat(orc.formatOptions().getString("compress", "")).isEqualTo("lz4"); + assertThat(orc.orcProperties().getProperty(IDENTIFIER + ".haha", "")).isEqualTo("1"); + assertThat(orc.orcProperties().getProperty(IDENTIFIER + ".compress", "")).isEqualTo("lz4"); } @Test @@ -42,7 +43,7 @@ public class OrcFileFormatTest { options.setString("haha", "1"); options.setString("compress", "zlib"); OrcFileFormat orc = new OrcFileFormatFactory().create(options); - assertThat(orc.formatOptions().getString("haha", "")).isEqualTo("1"); - assertThat(orc.formatOptions().getString("compress", "")).isEqualTo("zlib"); + assertThat(orc.orcProperties().getProperty(IDENTIFIER + ".haha", "")).isEqualTo("1"); + assertThat(orc.orcProperties().getProperty(IDENTIFIER + ".compress", "")).isEqualTo("zlib"); } }
