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/12/01 07:02:58 UTC
(incubator-paimon) 24/46: [spark] override equals as a preparative step for performance optimization (#2411)
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch release-0.6
in repository https://gitbox.apache.org/repos/asf/incubator-paimon.git
commit 25fe0dbf477b392c5246da41e5f6e581d68244fa
Author: Yann Byron <bi...@gmail.com>
AuthorDate: Wed Nov 29 10:14:13 2023 +0800
[spark] override equals as a preparative step for performance optimization (#2411)
---
.../apache/paimon/spark/SparkInputPartition.java | 14 +++++++++++
.../apache/paimon/spark/SparkReaderFactory.java | 14 +++++++++++
.../java/org/apache/paimon/spark/SparkScan.java | 16 +------------
.../org/apache/paimon/spark/PaimonBatch.scala} | 28 ++++++++++++----------
4 files changed, 44 insertions(+), 28 deletions(-)
diff --git a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkInputPartition.java b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkInputPartition.java
index 5c22a26e0..47f32ab42 100644
--- a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkInputPartition.java
+++ b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkInputPartition.java
@@ -36,4 +36,18 @@ public class SparkInputPartition implements InputPartition {
public Split split() {
return split;
}
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+
+ SparkInputPartition that = (SparkInputPartition) o;
+ return this.split.equals(that.split);
+ }
}
diff --git a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkReaderFactory.java b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkReaderFactory.java
index 279aa5853..986183832 100644
--- a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkReaderFactory.java
+++ b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkReaderFactory.java
@@ -61,4 +61,18 @@ public class SparkReaderFactory implements PartitionReaderFactory {
SparkInternalRow row = new SparkInternalRow(readBuilder.readType());
return new SparkInputPartitionReader(ioManager, iterator, row);
}
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+
+ SparkReaderFactory that = (SparkReaderFactory) o;
+ return this.readBuilder.equals(that.readBuilder);
+ }
}
diff --git a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkScan.java b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkScan.java
index 04d959da6..891420beb 100644
--- a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkScan.java
+++ b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkScan.java
@@ -25,8 +25,6 @@ import org.apache.paimon.table.source.ReadBuilder;
import org.apache.paimon.table.source.Split;
import org.apache.spark.sql.connector.read.Batch;
-import org.apache.spark.sql.connector.read.InputPartition;
-import org.apache.spark.sql.connector.read.PartitionReaderFactory;
import org.apache.spark.sql.connector.read.Scan;
import org.apache.spark.sql.connector.read.Statistics;
import org.apache.spark.sql.connector.read.SupportsReportStatistics;
@@ -66,19 +64,7 @@ public class SparkScan implements Scan, SupportsReportStatistics {
@Override
public Batch toBatch() {
- return new Batch() {
- @Override
- public InputPartition[] planInputPartitions() {
- return splits().stream()
- .map(SparkInputPartition::new)
- .toArray(InputPartition[]::new);
- }
-
- @Override
- public PartitionReaderFactory createReaderFactory() {
- return new SparkReaderFactory(readBuilder);
- }
- };
+ return new PaimonBatch(splits().toArray(new Split[0]), readBuilder);
}
@Override
diff --git a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkInputPartition.java b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonBatch.scala
similarity index 51%
copy from paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkInputPartition.java
copy to paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonBatch.scala
index 5c22a26e0..b356cf90e 100644
--- a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkInputPartition.java
+++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonBatch.scala
@@ -15,25 +15,27 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+package org.apache.paimon.spark
-package org.apache.paimon.spark;
+import org.apache.paimon.table.source.{ReadBuilder, Split}
-import org.apache.paimon.table.source.Split;
+import org.apache.spark.sql.connector.read.{Batch, InputPartition, PartitionReaderFactory}
-import org.apache.spark.sql.connector.read.InputPartition;
+/** A Spark {@link Batch} for paimon. */
+case class PaimonBatch(splits: Array[Split], readBuilder: ReadBuilder) extends Batch {
-/** A Spark {@link InputPartition} for paimon. */
-public class SparkInputPartition implements InputPartition {
+ override def planInputPartitions(): Array[InputPartition] =
+ splits.map(new SparkInputPartition(_).asInstanceOf[InputPartition])
- private static final long serialVersionUID = 1L;
+ override def createReaderFactory(): PartitionReaderFactory = new SparkReaderFactory(readBuilder)
- private final Split split;
+ override def equals(obj: Any): Boolean = {
+ obj match {
+ case other: PaimonBatch =>
+ this.splits.sameElements(other.splits) &&
+ readBuilder.equals(other.readBuilder)
- public SparkInputPartition(Split split) {
- this.split = split;
- }
-
- public Split split() {
- return split;
+ case _ => false
}
+ }
}