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 bee050f0 [FLINK-29345] Create reusing reader/writer config in orc
format
bee050f0 is described below
commit bee050f0301b00c620d4129c7a74ffbbc3d628ec
Author: shammon <[email protected]>
AuthorDate: Sat Oct 8 15:45:00 2022 +0800
[FLINK-29345] Create reusing reader/writer config in orc format
This closes #309
---
.../table/store/format/orc/OrcFileFormat.java | 38 ++++++++++++++--------
.../table/store/format/orc/OrcFileFormatTest.java | 9 ++---
2 files changed, 30 insertions(+), 17 deletions(-)
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 27b2f994..b7e8ce39 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,9 @@ 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.vector.Vectorizer;
import org.apache.flink.orc.writer.OrcBulkWriterFactory;
+import org.apache.flink.orc.writer.ThreadLocalClassLoaderConfiguration;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.store.file.predicate.Predicate;
@@ -54,16 +56,21 @@ 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()));
+ this.writerConf = new org.apache.hadoop.conf.Configuration();
}
@VisibleForTesting
- Configuration formatOptions() {
- return formatOptions;
+ Properties orcProperties() {
+ return orcProperties;
}
@Override
@@ -85,28 +92,33 @@ 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,
+ readerConf,
(RowType) refineLogicalType(type),
Projection.of(projection).toTopLevelIndexes(),
orcPredicates);
}
+ /**
+ * The {@link OrcBulkWriterFactory} will create {@link
ThreadLocalClassLoaderConfiguration} from
+ * the input writer config to avoid classloader leaks.
+ *
+ * <p>TODO: The {@link ThreadLocalClassLoaderConfiguration} in {@link
OrcBulkWriterFactory}
+ * should be removed after https://issues.apache.org/jira/browse/ORC-653
is fixed.
+ *
+ * @param type The data type for the {@link BulkWriter}
+ * @return The factory of the {@link BulkWriter}
+ */
@Override
public BulkWriter.Factory<RowData> createWriterFactory(RowType type) {
LogicalType[] orcTypes = type.getChildren().toArray(new
LogicalType[0]);
TypeDescription typeDescription =
OrcSplitReaderUtil.logicalTypeToOrcType(refineLogicalType(type));
+ Vectorizer<RowData> vectorizer =
+ new RowDataVectorizer(typeDescription.toString(), orcTypes);
- return new OrcBulkWriterFactory<>(
- new RowDataVectorizer(typeDescription.toString(), orcTypes),
- getOrcProperties(formatOptions),
- new org.apache.hadoop.conf.Configuration());
+ return new OrcBulkWriterFactory<>(vectorizer, orcProperties,
writerConf);
}
private static Properties getOrcProperties(ReadableConfig options) {
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");
}
}
