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 5fbd22e [FLINK-27070] Reuse FileFormat instead of
DecodingFormat/EncodingFormat to ensure thread safety
5fbd22e is described below
commit 5fbd22e262eb05cee5ddb1d19a30b46517d551ce
Author: tsreaper <[email protected]>
AuthorDate: Wed Apr 6 16:06:30 2022 +0800
[FLINK-27070] Reuse FileFormat instead of DecodingFormat/EncodingFormat to
ensure thread safety
This closes #74
---
.../flink/table/store/file/format/FileFormat.java | 6 +++++-
.../table/store/file/format/FileFormatImpl.java | 20 ++++++++++----------
.../table/store/file/manifest/ManifestFile.java | 12 +++++-------
.../table/store/file/manifest/ManifestList.java | 12 +++++-------
.../store/file/mergetree/sst/SstFileReader.java | 11 ++++-------
.../store/file/mergetree/sst/SstFileWriter.java | 12 +++++-------
.../store/file/operation/FileStoreWriteImpl.java | 8 +++-----
.../table/store/format/avro/AvroFileFormat.java | 8 ++++----
.../flink/table/store/format/orc/OrcFileFormat.java | 8 ++++----
9 files changed, 45 insertions(+), 52 deletions(-)
diff --git
a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/format/FileFormat.java
b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/format/FileFormat.java
index 76c9cf9..6bf7d23 100644
---
a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/format/FileFormat.java
+++
b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/format/FileFormat.java
@@ -47,7 +47,11 @@ import java.util.ServiceLoader;
import static
org.apache.flink.table.types.utils.TypeConversions.fromLogicalToDataType;
-/** Factory class which creates reader and writer factories for specific file
format. */
+/**
+ * Factory class which creates reader and writer factories for specific file
format.
+ *
+ * <p>NOTE: This class must be thread safe.
+ */
public abstract class FileFormat {
protected abstract BulkDecodingFormat<RowData> getDecodingFormat();
diff --git
a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/format/FileFormatImpl.java
b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/format/FileFormatImpl.java
index 0f8777c..35ba2fb 100644
---
a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/format/FileFormatImpl.java
+++
b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/format/FileFormatImpl.java
@@ -30,27 +30,27 @@ import org.apache.flink.table.factories.FactoryUtil;
/** A {@link FileFormat} which discovers reader and writer from format
identifier. */
public class FileFormatImpl extends FileFormat {
- private final ClassLoader classLoader;
- private final String formatIdentifier;
+ private final BulkReaderFormatFactory readerFactory;
+ private final BulkWriterFormatFactory writerFactory;
private final ReadableConfig formatOptions;
public FileFormatImpl(
ClassLoader classLoader, String formatIdentifier, ReadableConfig
formatOptions) {
- this.classLoader = classLoader;
- this.formatIdentifier = formatIdentifier;
+ this.readerFactory =
+ FactoryUtil.discoverFactory(
+ classLoader, BulkReaderFormatFactory.class,
formatIdentifier);
+ this.writerFactory =
+ FactoryUtil.discoverFactory(
+ classLoader, BulkWriterFormatFactory.class,
formatIdentifier);
this.formatOptions = formatOptions;
}
protected BulkDecodingFormat<RowData> getDecodingFormat() {
- return FactoryUtil.discoverFactory(
- classLoader, BulkReaderFormatFactory.class,
formatIdentifier)
- .createDecodingFormat(null, formatOptions); // context is
useless
+ return readerFactory.createDecodingFormat(null, formatOptions); //
context is useless
}
@Override
protected EncodingFormat<BulkWriter.Factory<RowData>> getEncodingFormat() {
- return FactoryUtil.discoverFactory(
- classLoader, BulkWriterFormatFactory.class,
formatIdentifier)
- .createEncodingFormat(null, formatOptions); // context is
useless
+ return writerFactory.createEncodingFormat(null, formatOptions); //
context is useless
}
}
diff --git
a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/manifest/ManifestFile.java
b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/manifest/ManifestFile.java
index 49f69fe..0fa2132 100644
---
a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/manifest/ManifestFile.java
+++
b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/manifest/ManifestFile.java
@@ -173,8 +173,7 @@ public class ManifestFile {
private final RowType partitionType;
private final RowType keyType;
private final RowType valueType;
- private final BulkFormat<RowData, FileSourceSplit> readerFactory;
- private final BulkWriter.Factory<RowData> writerFactory;
+ private final FileFormat fileFormat;
private final FileStorePathFactory pathFactory;
private final long suggestedFileSize;
@@ -188,19 +187,18 @@ public class ManifestFile {
this.partitionType = partitionType;
this.keyType = keyType;
this.valueType = valueType;
- RowType entryType = ManifestEntry.schema(partitionType, keyType,
valueType);
- this.readerFactory = fileFormat.createReaderFactory(entryType);
- this.writerFactory = fileFormat.createWriterFactory(entryType);
+ this.fileFormat = fileFormat;
this.pathFactory = pathFactory;
this.suggestedFileSize = suggestedFileSize;
}
public ManifestFile create() {
+ RowType entryType = ManifestEntry.schema(partitionType, keyType,
valueType);
return new ManifestFile(
partitionType,
new ManifestEntrySerializer(partitionType, keyType,
valueType),
- readerFactory,
- writerFactory,
+ fileFormat.createReaderFactory(entryType),
+ fileFormat.createWriterFactory(entryType),
pathFactory,
suggestedFileSize);
}
diff --git
a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/manifest/ManifestList.java
b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/manifest/ManifestList.java
index e15f546..82b64c0 100644
---
a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/manifest/ManifestList.java
+++
b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/manifest/ManifestList.java
@@ -103,24 +103,22 @@ public class ManifestList {
public static class Factory {
private final RowType partitionType;
- private final BulkFormat<RowData, FileSourceSplit> readerFactory;
- private final BulkWriter.Factory<RowData> writerFactory;
+ private final FileFormat fileFormat;
private final FileStorePathFactory pathFactory;
public Factory(
RowType partitionType, FileFormat fileFormat,
FileStorePathFactory pathFactory) {
this.partitionType = partitionType;
- RowType metaType = ManifestFileMeta.schema(partitionType);
- this.readerFactory = fileFormat.createReaderFactory(metaType);
- this.writerFactory = fileFormat.createWriterFactory(metaType);
+ this.fileFormat = fileFormat;
this.pathFactory = pathFactory;
}
public ManifestList create() {
+ RowType metaType = ManifestFileMeta.schema(partitionType);
return new ManifestList(
new ManifestFileMetaSerializer(partitionType),
- readerFactory,
- writerFactory,
+ fileFormat.createReaderFactory(metaType),
+ fileFormat.createWriterFactory(metaType),
pathFactory);
}
}
diff --git
a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/sst/SstFileReader.java
b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/sst/SstFileReader.java
index b3c0a59..4f56bf8 100644
---
a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/sst/SstFileReader.java
+++
b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/sst/SstFileReader.java
@@ -125,7 +125,6 @@ public class SstFileReader {
private int[][] valueProjection;
private RowType projectedKeyType;
private RowType projectedValueType;
- private BulkFormat<RowData, FileSourceSplit> readerFactory;
public Factory(
RowType keyType,
@@ -155,21 +154,19 @@ public class SstFileReader {
}
public SstFileReader create(BinaryRowData partition, int bucket) {
+ RowType recordType = KeyValue.schema(keyType, valueType);
+ int[][] projection =
+ KeyValue.project(keyProjection, valueProjection,
keyType.getFieldCount());
return new SstFileReader(
projectedKeyType,
projectedValueType,
- readerFactory,
+ fileFormat.createReaderFactory(recordType, projection),
pathFactory.createSstPathFactory(partition, bucket));
}
private void applyProjection() {
projectedKeyType = (RowType)
Projection.of(keyProjection).project(keyType);
projectedValueType = (RowType)
Projection.of(valueProjection).project(valueType);
-
- RowType recordType = KeyValue.schema(keyType, valueType);
- int[][] projection =
- KeyValue.project(keyProjection, valueProjection,
keyType.getFieldCount());
- readerFactory = fileFormat.createReaderFactory(recordType,
projection);
}
}
}
diff --git
a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/sst/SstFileWriter.java
b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/sst/SstFileWriter.java
index afc7e19..c72826a 100644
---
a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/sst/SstFileWriter.java
+++
b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/sst/SstFileWriter.java
@@ -265,8 +265,7 @@ public class SstFileWriter {
private final RowType keyType;
private final RowType valueType;
- private final BulkWriter.Factory<RowData> writerFactory;
- private final FileStatsExtractor fileStatsExtractor;
+ private final FileFormat fileFormat;
private final FileStorePathFactory pathFactory;
private final long suggestedFileSize;
@@ -278,19 +277,18 @@ public class SstFileWriter {
long suggestedFileSize) {
this.keyType = keyType;
this.valueType = valueType;
- RowType recordType = KeyValue.schema(keyType, valueType);
- this.writerFactory = fileFormat.createWriterFactory(recordType);
- this.fileStatsExtractor =
fileFormat.createStatsExtractor(recordType).orElse(null);
+ this.fileFormat = fileFormat;
this.pathFactory = pathFactory;
this.suggestedFileSize = suggestedFileSize;
}
public SstFileWriter create(BinaryRowData partition, int bucket) {
+ RowType recordType = KeyValue.schema(keyType, valueType);
return new SstFileWriter(
keyType,
valueType,
- writerFactory,
- fileStatsExtractor,
+ fileFormat.createWriterFactory(recordType),
+ fileFormat.createStatsExtractor(recordType).orElse(null),
pathFactory.createSstPathFactory(partition, bucket),
suggestedFileSize);
}
diff --git
a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/operation/FileStoreWriteImpl.java
b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/operation/FileStoreWriteImpl.java
index 2a47af1..a313c68 100644
---
a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/operation/FileStoreWriteImpl.java
+++
b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/operation/FileStoreWriteImpl.java
@@ -119,7 +119,7 @@ public class FileStoreWriteImpl implements FileStoreWrite {
sstFileWriter.valueType(),
options.writeBufferSize,
options.pageSize),
- createCompactManager(partition, bucket, sstFileWriter,
compactExecutor),
+ createCompactManager(partition, bucket, compactExecutor),
new Levels(keyComparator, restoreFiles, options.numLevels),
maxSequenceNumber,
keyComparator,
@@ -129,15 +129,13 @@ public class FileStoreWriteImpl implements FileStoreWrite
{
}
private CompactManager createCompactManager(
- BinaryRowData partition,
- int bucket,
- SstFileWriter sstFileWriter,
- ExecutorService compactExecutor) {
+ BinaryRowData partition, int bucket, ExecutorService
compactExecutor) {
CompactStrategy compactStrategy =
new UniversalCompaction(
options.maxSizeAmplificationPercent,
options.sizeRatio,
options.numSortedRunMax);
+ SstFileWriter sstFileWriter = sstFileWriterFactory.create(partition,
bucket);
CompactManager.Rewriter rewriter =
(outputLevel, dropDelete, sections) ->
sstFileWriter.write(
diff --git
a/flink-table-store-format/src/main/java/org/apache/flink/table/store/format/avro/AvroFileFormat.java
b/flink-table-store-format/src/main/java/org/apache/flink/table/store/format/avro/AvroFileFormat.java
index 10a6fa9..fcaed5c 100644
---
a/flink-table-store-format/src/main/java/org/apache/flink/table/store/format/avro/AvroFileFormat.java
+++
b/flink-table-store-format/src/main/java/org/apache/flink/table/store/format/avro/AvroFileFormat.java
@@ -28,21 +28,21 @@ import org.apache.flink.table.store.file.format.FileFormat;
/** Avro {@link FileFormat}. */
public class AvroFileFormat extends FileFormat {
+ private final org.apache.flink.formats.avro.AvroFileFormatFactory factory;
private final ReadableConfig formatOptions;
public AvroFileFormat(ReadableConfig formatOptions) {
+ this.factory = new
org.apache.flink.formats.avro.AvroFileFormatFactory();
this.formatOptions = formatOptions;
}
@Override
protected BulkDecodingFormat<RowData> getDecodingFormat() {
- return new org.apache.flink.formats.avro.AvroFileFormatFactory()
- .createDecodingFormat(null, formatOptions); // context is
useless
+ return factory.createDecodingFormat(null, formatOptions); // context
is useless
}
@Override
protected EncodingFormat<BulkWriter.Factory<RowData>> getEncodingFormat() {
- return new org.apache.flink.formats.avro.AvroFileFormatFactory()
- .createEncodingFormat(null, formatOptions); // context is
useless
+ return factory.createEncodingFormat(null, formatOptions); // context
is useless
}
}
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 ddb6a97..e56eb05 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
@@ -32,22 +32,22 @@ import java.util.Optional;
/** Orc {@link FileFormat}. */
public class OrcFileFormat extends FileFormat {
+ private final org.apache.flink.orc.OrcFileFormatFactory factory;
private final ReadableConfig formatOptions;
public OrcFileFormat(ReadableConfig formatOptions) {
+ this.factory = new org.apache.flink.orc.OrcFileFormatFactory();
this.formatOptions = formatOptions;
}
@Override
protected BulkDecodingFormat<RowData> getDecodingFormat() {
- return new org.apache.flink.orc.OrcFileFormatFactory()
- .createDecodingFormat(null, formatOptions); // context is
useless
+ return factory.createDecodingFormat(null, formatOptions); // context
is useless
}
@Override
protected EncodingFormat<BulkWriter.Factory<RowData>> getEncodingFormat() {
- return new org.apache.flink.orc.OrcFileFormatFactory()
- .createEncodingFormat(null, formatOptions); // context is
useless
+ return factory.createEncodingFormat(null, formatOptions); // context
is useless
}
@Override
