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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
