This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink-table-store.git
The following commit(s) were added to refs/heads/master by this push:
new 835632c6 [FLINK-29345] Create reusing reader/writer config in orc
format
835632c6 is described below
commit 835632c6e4758ad7d11ccbdb3a8ebb8dfa6aa709
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 8b028210..9910409e 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 javax.annotation.Nullable;
@@ -50,16 +52,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
@@ -81,12 +91,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
@@ -94,11 +100,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");
}
}
