JingsongLi commented on code in PR #2367: URL: https://github.com/apache/incubator-paimon/pull/2367#discussion_r1409055612
########## paimon-spark/paimon-spark-3.1/src/main/scala/org/apache/paimon/spark/SparkScanBuilder.scala: ########## @@ -0,0 +1,107 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.paimon.spark + +import org.apache.paimon.predicate.{PartitionPredicateVisitor, Predicate, PredicateBuilder} +import org.apache.paimon.table.Table + +import org.apache.spark.internal.Logging +import org.apache.spark.sql.connector.read.{Scan, ScanBuilder, SupportsPushDownFilters, SupportsPushDownRequiredColumns} +import org.apache.spark.sql.sources.Filter +import org.apache.spark.sql.types.StructType + +import scala.collection.mutable + +class SparkScanBuilder(table: Table) Review Comment: Can we have a `SparkBaseScanBuilder` to reuse code? ########## paimon-core/src/main/java/org/apache/paimon/table/source/RawFile.java: ########## @@ -76,12 +80,18 @@ public long schemaId() { return schemaId; } + /** row count of the file. */ + public long rowCount() { + return rowCount; + } + public void serialize(DataOutputView out) throws IOException { out.writeUTF(path); out.writeLong(offset); out.writeLong(length); out.writeUTF(format); out.writeLong(schemaId); + out.writeLong(rowCount); Review Comment: Increment `DataSplit.serialVersionUID` too. ########## paimon-core/src/main/java/org/apache/paimon/table/source/InnerTableScanImpl.java: ########## @@ -68,9 +81,96 @@ public TableScan.Plan plan() { if (hasNext) { hasNext = false; StartingScanner.Result result = startingScanner.scan(snapshotReader); - return DataFilePlan.fromResult(result); + StartingScanner.Result limitedResult = applyPushDownLimit(result); + return DataFilePlan.fromResult(limitedResult); } else { throw new EndOfScanException(); } } + + private StartingScanner.Result applyPushDownLimit(StartingScanner.Result result) { + if (pushDownLimit != null && result instanceof StartingScanner.ScannedResult) { + long scannedRowCount = 0; + SnapshotReader.Plan plan = ((StartingScanner.ScannedResult) result).plan(); + List<DataSplit> splits = plan.dataSplits(); + List<DataSplit> limitedSplits = new ArrayList<>(); + for (int i = 0; i < splits.size(); i++) { + if (scannedRowCount >= pushDownLimit) { + break; + } + + DataSplit split = splits.get(i); + long splitRowCount = getRowCountForSplit(split); + if (scannedRowCount + splitRowCount <= pushDownLimit) { + limitedSplits.add(split); + scannedRowCount += splitRowCount; + } else { + DataSplit newSplit = composeDataSplit(split, pushDownLimit - scannedRowCount); Review Comment: I think we don't need to introduce `composeDataSplit`. The reason we introduce limit pushdown is to reduce split number. We don't need to reduce files in split. ########## paimon-spark/paimon-spark-3.1/src/main/scala/org/apache/paimon/spark/SparkScanBuilder.scala: ########## @@ -0,0 +1,107 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.paimon.spark + +import org.apache.paimon.predicate.{PartitionPredicateVisitor, Predicate, PredicateBuilder} +import org.apache.paimon.table.Table + +import org.apache.spark.internal.Logging +import org.apache.spark.sql.connector.read.{Scan, ScanBuilder, SupportsPushDownFilters, SupportsPushDownRequiredColumns} +import org.apache.spark.sql.sources.Filter +import org.apache.spark.sql.types.StructType + +import scala.collection.mutable + +class SparkScanBuilder(table: Table) Review Comment: You need to add class for spark-3.2 too? And add some itcase for spark-3.2 too (For ITCase, we also can have some abstraction to reduce code replication). ########## paimon-core/src/main/java/org/apache/paimon/table/source/InnerTableScanImpl.java: ########## @@ -68,9 +81,96 @@ public TableScan.Plan plan() { if (hasNext) { hasNext = false; StartingScanner.Result result = startingScanner.scan(snapshotReader); - return DataFilePlan.fromResult(result); + StartingScanner.Result limitedResult = applyPushDownLimit(result); + return DataFilePlan.fromResult(limitedResult); } else { throw new EndOfScanException(); } } + + private StartingScanner.Result applyPushDownLimit(StartingScanner.Result result) { + if (pushDownLimit != null && result instanceof StartingScanner.ScannedResult) { + long scannedRowCount = 0; + SnapshotReader.Plan plan = ((StartingScanner.ScannedResult) result).plan(); + List<DataSplit> splits = plan.dataSplits(); + List<DataSplit> limitedSplits = new ArrayList<>(); + for (int i = 0; i < splits.size(); i++) { + if (scannedRowCount >= pushDownLimit) { + break; + } + + DataSplit split = splits.get(i); + long splitRowCount = getRowCountForSplit(split); + if (scannedRowCount + splitRowCount <= pushDownLimit) { + limitedSplits.add(split); + scannedRowCount += splitRowCount; + } else { + DataSplit newSplit = composeDataSplit(split, pushDownLimit - scannedRowCount); + limitedSplits.add(newSplit); + scannedRowCount += getRowCountForSplit(newSplit); + } + } + + SnapshotReader.Plan newPlan = + new SnapshotReader.Plan() { + @Nullable + @Override + public Long watermark() { + return plan.watermark(); + } + + @Nullable + @Override + public Long snapshotId() { + return plan.snapshotId(); + } + + @Override + public List<Split> splits() { + return (List) limitedSplits; + } + }; + return new StartingScanner.ScannedResult(newPlan); + } else { + return result; + } + } + + /** + * 0 represents that we can't compute the row count of this split, 'cause this split needs to be + * merged. + */ + private long getRowCountForSplit(DataSplit split) { + if (split.convertToRawFiles().isPresent()) { + return split.convertToRawFiles().get().stream() + .map(RawFile::rowCount) + .reduce(Long::sum) + .orElse(0L); + } else { + return 0L; + } + } + + private DataSplit composeDataSplit(DataSplit split, long rowCountRequired) { + List<DataFileMeta> dataFiles = new ArrayList<>(); + List<RawFile> rawFiles = new ArrayList<>(); + long scannedRowCount = 0; + + List<DataFileMeta> originalDataFiles = split.dataFiles(); + List<RawFile> originalRawFiles = split.convertToRawFiles().get(); Review Comment: Just get? Exception when it is empty? -- 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]
