You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@paimon.apache.org by "JingsongLi (via GitHub)" <gi...@apache.org> on 2023/05/05 04:25:00 UTC
[GitHub] [incubator-paimon] JingsongLi commented on a diff in pull request #1072: [core] limit parallelly read file memory usage, extract some methods
JingsongLi commented on code in PR #1072:
URL: https://github.com/apache/incubator-paimon/pull/1072#discussion_r1185683577
##########
paimon-core/src/main/java/org/apache/paimon/operation/AbstractFileStoreScan.java:
##########
@@ -195,6 +196,70 @@ public FileStoreScan withManifestCacheFilter(ManifestCacheFilter manifestFilter)
@Override
public Plan plan() {
+ Pair<Long, List<ManifestEntry>> planResult =
+ doPlan(
+ // how to process entry files
+ entries ->
+ entries.parallelStream()
+ .filter(this::filterManifestFileMeta)
+ .flatMap(m -> readManifestFileMeta(m).stream())
+ .filter(this::filterByStats)
+ .collect(Collectors.toList()),
+ // post process filter files
+ file -> {
+ if (checkNumOfBuckets && file.totalBuckets() != numOfBuckets) {
+ String partInfo =
+ partitionConverter.getArity() > 0
+ ? "partition "
+ + FileStorePathFactory.getPartitionComputer(
+ partitionConverter
+ .rowType(),
+ FileStorePathFactory
+ .PARTITION_DEFAULT_NAME
+ .defaultValue())
+ .generatePartValues(
+ file.partition())
+ : "table";
+ throw new RuntimeException(
+ String.format(
+ "Try to write %s with a new bucket num %d, but the previous bucket num is %d. "
+ + "Please switch to batch mode, and perform INSERT OVERWRITE to rescale current data layout first.",
+ partInfo, numOfBuckets, file.totalBuckets()));
+ }
+
+ // bucket filter should not be applied along with partition filter
+ // because the specifiedBucket is computed against the current
+ // numOfBuckets
+ // however entry.bucket() was computed against the old numOfBuckets
+ // and thus the filtered manifest entries might be empty
+ // which renders the bucket check invalid
+ if (filterByBucket(file)
Review Comment:
'if' statement can be simplified
##########
paimon-core/src/main/java/org/apache/paimon/operation/AbstractFileStoreScan.java:
##########
@@ -195,6 +196,70 @@ public FileStoreScan withManifestCacheFilter(ManifestCacheFilter manifestFilter)
@Override
public Plan plan() {
+ Pair<Long, List<ManifestEntry>> planResult =
+ doPlan(
+ // how to process entry files
+ entries ->
+ entries.parallelStream()
+ .filter(this::filterManifestFileMeta)
+ .flatMap(m -> readManifestFileMeta(m).stream())
+ .filter(this::filterByStats)
+ .collect(Collectors.toList()),
+ // post process filter files
+ file -> {
+ if (checkNumOfBuckets && file.totalBuckets() != numOfBuckets) {
+ String partInfo =
+ partitionConverter.getArity() > 0
+ ? "partition "
+ + FileStorePathFactory.getPartitionComputer(
+ partitionConverter
+ .rowType(),
+ FileStorePathFactory
+ .PARTITION_DEFAULT_NAME
+ .defaultValue())
+ .generatePartValues(
+ file.partition())
+ : "table";
+ throw new RuntimeException(
+ String.format(
+ "Try to write %s with a new bucket num %d, but the previous bucket num is %d. "
+ + "Please switch to batch mode, and perform INSERT OVERWRITE to rescale current data layout first.",
+ partInfo, numOfBuckets, file.totalBuckets()));
+ }
+
+ // bucket filter should not be applied along with partition filter
+ // because the specifiedBucket is computed against the current
+ // numOfBuckets
+ // however entry.bucket() was computed against the old numOfBuckets
+ // and thus the filtered manifest entries might be empty
+ // which renders the bucket check invalid
+ if (filterByBucket(file)
+ && filterByBucketSelector(file)
+ && filterByLevel(file)) {
+ return true;
+ }
+ return false;
+ });
+ final Long readSnapshotId = planResult.getLeft();
+ final List<ManifestEntry> files = planResult.getRight();
+
+ return new Plan() {
+ @Nullable
+ @Override
+ public Long snapshotId() {
+ return readSnapshotId;
+ }
+
+ @Override
+ public List<ManifestEntry> files() {
+ return files;
+ }
+ };
+ }
+
+ private <T extends AbstractManifestEntry> Pair<Long, List<T>> doPlan(
+ Function<List<ManifestFileMeta>, List<T>> processor,
Review Comment:
`readManifestFile`?
##########
paimon-core/src/main/java/org/apache/paimon/operation/AbstractFileStoreScan.java:
##########
@@ -195,6 +196,70 @@ public FileStoreScan withManifestCacheFilter(ManifestCacheFilter manifestFilter)
@Override
public Plan plan() {
+ Pair<Long, List<ManifestEntry>> planResult =
+ doPlan(
+ // how to process entry files
+ entries ->
+ entries.parallelStream()
+ .filter(this::filterManifestFileMeta)
+ .flatMap(m -> readManifestFileMeta(m).stream())
+ .filter(this::filterByStats)
+ .collect(Collectors.toList()),
+ // post process filter files
+ file -> {
+ if (checkNumOfBuckets && file.totalBuckets() != numOfBuckets) {
+ String partInfo =
+ partitionConverter.getArity() > 0
+ ? "partition "
+ + FileStorePathFactory.getPartitionComputer(
+ partitionConverter
+ .rowType(),
+ FileStorePathFactory
+ .PARTITION_DEFAULT_NAME
+ .defaultValue())
+ .generatePartValues(
+ file.partition())
+ : "table";
+ throw new RuntimeException(
+ String.format(
+ "Try to write %s with a new bucket num %d, but the previous bucket num is %d. "
+ + "Please switch to batch mode, and perform INSERT OVERWRITE to rescale current data layout first.",
+ partInfo, numOfBuckets, file.totalBuckets()));
+ }
+
+ // bucket filter should not be applied along with partition filter
+ // because the specifiedBucket is computed against the current
+ // numOfBuckets
+ // however entry.bucket() was computed against the old numOfBuckets
+ // and thus the filtered manifest entries might be empty
+ // which renders the bucket check invalid
+ if (filterByBucket(file)
+ && filterByBucketSelector(file)
+ && filterByLevel(file)) {
+ return true;
+ }
+ return false;
+ });
+ final Long readSnapshotId = planResult.getLeft();
+ final List<ManifestEntry> files = planResult.getRight();
+
+ return new Plan() {
+ @Nullable
+ @Override
+ public Long snapshotId() {
+ return readSnapshotId;
+ }
+
+ @Override
+ public List<ManifestEntry> files() {
+ return files;
+ }
+ };
+ }
+
+ private <T extends AbstractManifestEntry> Pair<Long, List<T>> doPlan(
+ Function<List<ManifestFileMeta>, List<T>> processor,
+ Function<T, Boolean> postFilterProcessor) {
Review Comment:
Maybe we don't need this one, postFilter can be applied to `SimpleManifestEntry` too.
If there is level filter, throw exception.
##########
paimon-core/src/main/java/org/apache/paimon/operation/AbstractFileStoreScan.java:
##########
@@ -195,6 +196,70 @@ public FileStoreScan withManifestCacheFilter(ManifestCacheFilter manifestFilter)
@Override
public Plan plan() {
+ Pair<Long, List<ManifestEntry>> planResult =
+ doPlan(
+ // how to process entry files
+ entries ->
+ entries.parallelStream()
Review Comment:
Can we also put these logical into `doPlan`?
Only `filterByStats` is special, we can just `instanceof ManifestEntry` in `filterByStats`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org