You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@paimon.apache.org by lz...@apache.org on 2023/06/20 02:36:28 UTC
[incubator-paimon] 05/09: [parquet] Fix connection leak in parquet ParquetUtil extractColumnStats (#1283)
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch release-0.4
in repository https://gitbox.apache.org/repos/asf/incubator-paimon.git
commit bd4a7deaa7ef85434e78693c429776245f0e52cd
Author: Jingsong Lee <ji...@gmail.com>
AuthorDate: Fri Jun 2 15:31:19 2023 +0800
[parquet] Fix connection leak in parquet ParquetUtil extractColumnStats (#1283)
---
.../paimon/table/FileStoreTableTestBase.java | 30 ++++++++++++++++++-
.../apache/paimon/format/parquet/ParquetUtil.java | 34 ++++++++++++----------
2 files changed, 47 insertions(+), 17 deletions(-)
diff --git a/paimon-core/src/test/java/org/apache/paimon/table/FileStoreTableTestBase.java b/paimon-core/src/test/java/org/apache/paimon/table/FileStoreTableTestBase.java
index d23132239..76ffb0848 100644
--- a/paimon-core/src/test/java/org/apache/paimon/table/FileStoreTableTestBase.java
+++ b/paimon-core/src/test/java/org/apache/paimon/table/FileStoreTableTestBase.java
@@ -67,6 +67,7 @@ import org.junit.jupiter.api.io.TempDir;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.Arguments;
import org.junit.jupiter.params.provider.MethodSource;
+import org.junit.jupiter.params.provider.ValueSource;
import java.io.IOException;
import java.util.ArrayList;
@@ -191,7 +192,7 @@ public abstract class FileStoreTableTestBase {
table =
createFileStoreTable(
- conf -> conf.set(FILE_FORMAT, CoreOptions.FileFormatType.AVRO));
+ conf -> conf.set(FILE_FORMAT, CoreOptions.FileFormatType.PARQUET));
write = table.newWrite(commitUser);
commit = table.newCommit(commitUser);
write.write(rowData(1, 11, 111L));
@@ -212,6 +213,33 @@ public abstract class FileStoreTableTestBase {
"2|22|222|binary|varbinary|mapKey:mapVal|multiset");
}
+ @ParameterizedTest(name = "{0}")
+ @ValueSource(strings = {"avro", "orc", "parquet"})
+ public void testMultipleCommits(String format) throws Exception {
+ FileStoreTable table =
+ createFileStoreTable(conf -> conf.setString(FILE_FORMAT.key(), format));
+ StreamTableWrite write = table.newWrite(commitUser);
+ StreamTableCommit commit = table.newCommit(commitUser);
+
+ List<String> expected = new ArrayList<>();
+ for (int i = 0; i < 60; i++) {
+ write.write(rowData(1, i, (long) i * 100));
+ commit.commit(i, write.prepareCommit(true, i));
+ expected.add(
+ String.format("1|%s|%s|binary|varbinary|mapKey:mapVal|multiset", i, i * 100));
+ }
+
+ write.close();
+ commit.close();
+
+ assertThat(
+ getResult(
+ table.newRead(),
+ toSplits(table.newSnapshotSplitReader().splits()),
+ BATCH_ROW_TO_STRING))
+ .containsExactlyElementsOf(expected);
+ }
+
@ParameterizedTest(name = "dynamic = {0}, partition={2}")
@MethodSource("overwriteTestData")
public void testOverwriteNothing(
diff --git a/paimon-format/src/main/java/org/apache/paimon/format/parquet/ParquetUtil.java b/paimon-format/src/main/java/org/apache/paimon/format/parquet/ParquetUtil.java
index 0732e722c..b7aa5d956 100644
--- a/paimon-format/src/main/java/org/apache/paimon/format/parquet/ParquetUtil.java
+++ b/paimon-format/src/main/java/org/apache/paimon/format/parquet/ParquetUtil.java
@@ -46,25 +46,27 @@ public class ParquetUtil {
*/
public static Map<String, Statistics<?>> extractColumnStats(FileIO fileIO, Path path)
throws IOException {
- ParquetMetadata parquetMetadata = getParquetReader(fileIO, path).getFooter();
- List<BlockMetaData> blockMetaDataList = parquetMetadata.getBlocks();
- Map<String, Statistics<?>> resultStats = new HashMap<>();
- for (BlockMetaData blockMetaData : blockMetaDataList) {
- List<ColumnChunkMetaData> columnChunkMetaDataList = blockMetaData.getColumns();
- for (ColumnChunkMetaData columnChunkMetaData : columnChunkMetaDataList) {
- Statistics<?> stats = columnChunkMetaData.getStatistics();
- String columnName = columnChunkMetaData.getPath().toDotString();
- Statistics<?> midStats;
- if (!resultStats.containsKey(columnName)) {
- midStats = stats;
- } else {
- midStats = resultStats.get(columnName);
- midStats.mergeStatistics(stats);
+ try (ParquetFileReader reader = getParquetReader(fileIO, path)) {
+ ParquetMetadata parquetMetadata = reader.getFooter();
+ List<BlockMetaData> blockMetaDataList = parquetMetadata.getBlocks();
+ Map<String, Statistics<?>> resultStats = new HashMap<>();
+ for (BlockMetaData blockMetaData : blockMetaDataList) {
+ List<ColumnChunkMetaData> columnChunkMetaDataList = blockMetaData.getColumns();
+ for (ColumnChunkMetaData columnChunkMetaData : columnChunkMetaDataList) {
+ Statistics<?> stats = columnChunkMetaData.getStatistics();
+ String columnName = columnChunkMetaData.getPath().toDotString();
+ Statistics<?> midStats;
+ if (!resultStats.containsKey(columnName)) {
+ midStats = stats;
+ } else {
+ midStats = resultStats.get(columnName);
+ midStats.mergeStatistics(stats);
+ }
+ resultStats.put(columnName, midStats);
}
- resultStats.put(columnName, midStats);
}
+ return resultStats;
}
- return resultStats;
}
/**