(incubator-paimon) branch master updated: [spark] define PaimonStatistics for CBO optimization (#2479)

[lzljs3620320]

This is an automated email from the ASF dual-hosted git repository.

lzljs3620320 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-paimon.git


The following commit(s) were added to refs/heads/master by this push:
     new 117bcb030 [spark] define PaimonStatistics for CBO optimization (#2479)
117bcb030 is described below

commit 117bcb0306dc39d9385e68bea8f8e6302c55fa46
Author: Yann Byron <biyan900...@gmail.com>
AuthorDate: Mon Dec 11 17:38:24 2023 +0800

    [spark] define PaimonStatistics for CBO optimization (#2479)
---
 .../org/apache/paimon/spark/PaimonBaseScan.scala   | 11 ++-----
 .../org/apache/paimon/spark/PaimonStatistics.scala | 35 ++++++++++++++++++++++
 2 files changed, 37 insertions(+), 9 deletions(-)

diff --git 
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonBaseScan.scala
 
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonBaseScan.scala
index 8eb732a99..d954c9e14 100644
--- 
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonBaseScan.scala
+++ 
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonBaseScan.scala
@@ -50,17 +50,10 @@ abstract class PaimonBaseScan(table: Table, readBuilder: 
ReadBuilder, desc: Stri
   }
 
   override def estimateStatistics(): Statistics = {
-    val rowCount = getSplits.map(_.rowCount).sum
-    val scannedTotalSize = rowCount * readSchema().defaultSize
-
-    new Statistics {
-      override def sizeInBytes(): OptionalLong = 
OptionalLong.of(scannedTotalSize)
-
-      override def numRows(): OptionalLong = OptionalLong.of(rowCount)
-    }
+    PaimonStatistics(this)
   }
 
-  private def getSplits: Array[Split] = {
+  def getSplits: Array[Split] = {
     if (splits == null) {
       splits = readBuilder.newScan().plan().splits().asScala.toArray
     }
diff --git 
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonStatistics.scala
 
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonStatistics.scala
new file mode 100644
index 000000000..208c5816c
--- /dev/null
+++ 
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonStatistics.scala
@@ -0,0 +1,35 @@
+/*
+ * 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.spark.sql.connector.read.Statistics
+
+import java.util.OptionalLong
+
+case class PaimonStatistics[T <: PaimonBaseScan](scan: T) extends Statistics {
+
+  private lazy val rowCount: Long = scan.getSplits.map(_.rowCount).sum
+
+  private lazy val scannedTotalSize: Long = rowCount * 
scan.readSchema().defaultSize
+
+  override def sizeInBytes(): OptionalLong = OptionalLong.of(scannedTotalSize)
+
+  override def numRows(): OptionalLong = OptionalLong.of(rowCount)
+
+  // TODO: extend columnStats for CBO
+}