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 2024/09/25 04:34:09 UTC
(paimon) branch master updated: [spark] PaimonSplitScan supports column pruning and filter push down (#4217)
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/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 3e871954e [spark] PaimonSplitScan supports column pruning and filter push down (#4217)
3e871954e is described below
commit 3e871954ed305b14d704386a93fd1d4ec9764e8d
Author: Xiduo You <ul...@gmail.com>
AuthorDate: Wed Sep 25 12:34:04 2024 +0800
[spark] PaimonSplitScan supports column pruning and filter push down (#4217)
---
.../org/apache/paimon/table/KnownSplitsTable.java | 90 +++++++++++++++++
.../scala/org/apache/paimon/spark/PaimonScan.scala | 2 +-
.../scala/org/apache/paimon/spark/PaimonScan.scala | 2 +-
.../paimon/spark/ColumnPruningAndPushDown.scala | 73 ++++++++++++++
.../org/apache/paimon/spark/PaimonBaseScan.scala | 46 +--------
.../scala/org/apache/paimon/spark/PaimonScan.scala | 2 +-
.../org/apache/paimon/spark/PaimonSplitScan.scala | 31 ++++--
.../scala/org/apache/paimon/spark/SparkTable.scala | 9 +-
.../spark/commands/MergeIntoPaimonTable.scala | 2 +-
.../paimon/spark/commands/PaimonCommand.scala | 78 +++++++++------
.../spark/commands/UpdatePaimonTableCommand.scala | 11 +--
.../paimon/spark/sql/DeletionVectorTest.scala | 110 ++++++++++++++++-----
.../test/scala/org/apache/spark/paimon/Utils.scala | 4 +
13 files changed, 341 insertions(+), 119 deletions(-)
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/KnownSplitsTable.java b/paimon-core/src/main/java/org/apache/paimon/table/KnownSplitsTable.java
new file mode 100644
index 000000000..035e90992
--- /dev/null
+++ b/paimon-core/src/main/java/org/apache/paimon/table/KnownSplitsTable.java
@@ -0,0 +1,90 @@
+/*
+ * 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.table;
+
+import org.apache.paimon.table.source.DataSplit;
+import org.apache.paimon.table.source.InnerTableRead;
+import org.apache.paimon.table.source.InnerTableScan;
+import org.apache.paimon.types.RowType;
+
+import java.util.List;
+import java.util.Map;
+
+/**
+ * A table to hold some known data splits. For now, it is only used by internal for Spark engine.
+ */
+public class KnownSplitsTable implements ReadonlyTable {
+ private final InnerTable origin;
+ private final DataSplit[] splits;
+
+ KnownSplitsTable(InnerTable origin, DataSplit[] splits) {
+ this.origin = origin;
+ this.splits = splits;
+ }
+
+ public static KnownSplitsTable create(InnerTable origin, DataSplit[] splits) {
+ return new KnownSplitsTable(origin, splits);
+ }
+
+ public DataSplit[] splits() {
+ return splits;
+ }
+
+ @Override
+ public String name() {
+ return origin.name();
+ }
+
+ @Override
+ public RowType rowType() {
+ return origin.rowType();
+ }
+
+ @Override
+ public List<String> primaryKeys() {
+ return origin.primaryKeys();
+ }
+
+ @Override
+ public List<String> partitionKeys() {
+ return origin.partitionKeys();
+ }
+
+ @Override
+ public Map<String, String> options() {
+ return origin.options();
+ }
+
+ @Override
+ public InnerTableRead newRead() {
+ return origin.newRead();
+ }
+
+ // ===== unused method ===========================================
+
+ @Override
+ public InnerTableScan newScan() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Table copy(Map<String, String> dynamicOptions) {
+ throw new UnsupportedOperationException();
+ }
+}
diff --git a/paimon-spark/paimon-spark-3.1/src/main/scala/org/apache/paimon/spark/PaimonScan.scala b/paimon-spark/paimon-spark-3.1/src/main/scala/org/apache/paimon/spark/PaimonScan.scala
index 254c63679..931cb830b 100644
--- a/paimon-spark/paimon-spark-3.1/src/main/scala/org/apache/paimon/spark/PaimonScan.scala
+++ b/paimon-spark/paimon-spark-3.1/src/main/scala/org/apache/paimon/spark/PaimonScan.scala
@@ -29,6 +29,6 @@ case class PaimonScan(
requiredSchema: StructType,
filters: Seq[Predicate],
reservedFilters: Seq[Filter],
- pushDownLimit: Option[Int],
+ override val pushDownLimit: Option[Int],
disableBucketedScan: Boolean = false)
extends PaimonBaseScan(table, requiredSchema, filters, reservedFilters, pushDownLimit)
diff --git a/paimon-spark/paimon-spark-3.2/src/main/scala/org/apache/paimon/spark/PaimonScan.scala b/paimon-spark/paimon-spark-3.2/src/main/scala/org/apache/paimon/spark/PaimonScan.scala
index 361b1e7a7..92a719b3b 100644
--- a/paimon-spark/paimon-spark-3.2/src/main/scala/org/apache/paimon/spark/PaimonScan.scala
+++ b/paimon-spark/paimon-spark-3.2/src/main/scala/org/apache/paimon/spark/PaimonScan.scala
@@ -34,7 +34,7 @@ case class PaimonScan(
requiredSchema: StructType,
filters: Seq[Predicate],
reservedFilters: Seq[Filter],
- pushDownLimit: Option[Int],
+ override val pushDownLimit: Option[Int],
disableBucketedScan: Boolean = false)
extends PaimonBaseScan(table, requiredSchema, filters, reservedFilters, pushDownLimit)
with SupportsRuntimeFiltering {
diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/ColumnPruningAndPushDown.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/ColumnPruningAndPushDown.scala
new file mode 100644
index 000000000..c36c2fff2
--- /dev/null
+++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/ColumnPruningAndPushDown.scala
@@ -0,0 +1,73 @@
+/*
+ * 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.{Predicate, PredicateBuilder}
+import org.apache.paimon.spark.schema.PaimonMetadataColumn
+import org.apache.paimon.table.Table
+import org.apache.paimon.table.source.ReadBuilder
+import org.apache.paimon.types.RowType
+
+import org.apache.spark.sql.connector.read.Scan
+import org.apache.spark.sql.types.StructType
+
+trait ColumnPruningAndPushDown extends Scan {
+ def table: Table
+ def requiredSchema: StructType
+ def filters: Seq[Predicate]
+ def pushDownLimit: Option[Int] = None
+
+ val tableRowType: RowType = table.rowType
+ val tableSchema: StructType = SparkTypeUtils.fromPaimonRowType(tableRowType)
+
+ final def partitionType: StructType = {
+ SparkTypeUtils.toSparkPartitionType(table)
+ }
+
+ private[paimon] val (requiredTableFields, metadataFields) = {
+ val nameToField = tableSchema.map(field => (field.name, field)).toMap
+ val _tableFields = requiredSchema.flatMap(field => nameToField.get(field.name))
+ val _metadataFields =
+ requiredSchema
+ .filterNot(field => tableSchema.fieldNames.contains(field.name))
+ .filter(field => PaimonMetadataColumn.SUPPORTED_METADATA_COLUMNS.contains(field.name))
+ (_tableFields, _metadataFields)
+ }
+
+ lazy val readBuilder: ReadBuilder = {
+ val _readBuilder = table.newReadBuilder()
+ val projection =
+ requiredTableFields.map(field => tableSchema.fieldNames.indexOf(field.name)).toArray
+ _readBuilder.withProjection(projection)
+ if (filters.nonEmpty) {
+ val pushedPredicate = PredicateBuilder.and(filters: _*)
+ _readBuilder.withFilter(pushedPredicate)
+ }
+ pushDownLimit.foreach(_readBuilder.withLimit)
+ _readBuilder
+ }
+
+ final def metadataColumns: Seq[PaimonMetadataColumn] = {
+ metadataFields.map(field => PaimonMetadataColumn.get(field.name, partitionType))
+ }
+
+ override def readSchema(): StructType = {
+ StructType(requiredTableFields ++ metadataFields)
+ }
+}
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 fb494a492..7fbd286d1 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
@@ -20,14 +20,12 @@ package org.apache.paimon.spark
import org.apache.paimon.{stats, CoreOptions}
import org.apache.paimon.annotation.VisibleForTesting
-import org.apache.paimon.predicate.{Predicate, PredicateBuilder}
+import org.apache.paimon.predicate.Predicate
import org.apache.paimon.spark.metric.SparkMetricRegistry
-import org.apache.paimon.spark.schema.PaimonMetadataColumn
import org.apache.paimon.spark.sources.PaimonMicroBatchStream
import org.apache.paimon.spark.statistics.StatisticsHelper
import org.apache.paimon.table.{DataTable, FileStoreTable, Table}
-import org.apache.paimon.table.source.{InnerTableScan, ReadBuilder, Split}
-import org.apache.paimon.types.RowType
+import org.apache.paimon.table.source.{InnerTableScan, Split}
import org.apache.spark.sql.connector.metric.{CustomMetric, CustomTaskMetric}
import org.apache.spark.sql.connector.read.{Batch, Scan, Statistics, SupportsReportStatistics}
@@ -48,22 +46,9 @@ abstract class PaimonBaseScan(
extends Scan
with SupportsReportStatistics
with ScanHelper
+ with ColumnPruningAndPushDown
with StatisticsHelper {
- val tableRowType: RowType = table.rowType
-
- private lazy val tableSchema = SparkTypeUtils.fromPaimonRowType(tableRowType)
-
- private[paimon] val (requiredTableFields, metadataFields) = {
- val nameToField = tableSchema.map(field => (field.name, field)).toMap
- val _tableFields = requiredSchema.flatMap(field => nameToField.get(field.name))
- val _metadataFields =
- requiredSchema
- .filterNot(field => tableSchema.fieldNames.contains(field.name))
- .filter(field => PaimonMetadataColumn.SUPPORTED_METADATA_COLUMNS.contains(field.name))
- (_tableFields, _metadataFields)
- }
-
protected var runtimeFilters: Array[Filter] = Array.empty
protected var inputPartitions: Seq[PaimonInputPartition] = _
@@ -79,21 +64,6 @@ abstract class PaimonBaseScan(
StructType(tableSchema.filter(field => fieldNames.contains(field.name)))
}
- lazy val readBuilder: ReadBuilder = {
- val _readBuilder = table.newReadBuilder()
-
- val projection =
- requiredTableFields.map(field => tableSchema.fieldNames.indexOf(field.name)).toArray
- _readBuilder.withProjection(projection)
- if (filters.nonEmpty) {
- val pushedPredicate = PredicateBuilder.and(filters: _*)
- _readBuilder.withFilter(pushedPredicate)
- }
- pushDownLimit.foreach(_readBuilder.withLimit)
-
- _readBuilder
- }
-
@VisibleForTesting
def getOriginSplits: Array[Split] = {
readBuilder
@@ -113,17 +83,7 @@ abstract class PaimonBaseScan(
inputPartitions
}
- final def partitionType: StructType = {
- SparkTypeUtils.toSparkPartitionType(table)
- }
-
- override def readSchema(): StructType = {
- StructType(requiredTableFields ++ metadataFields)
- }
-
override def toBatch: Batch = {
- val metadataColumns =
- metadataFields.map(field => PaimonMetadataColumn.get(field.name, partitionType))
PaimonBatch(lazyInputPartitions, readBuilder, metadataColumns)
}
diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonScan.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonScan.scala
index 5f73ae553..2f1e6c53a 100644
--- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonScan.scala
+++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonScan.scala
@@ -36,7 +36,7 @@ case class PaimonScan(
requiredSchema: StructType,
filters: Seq[Predicate],
reservedFilters: Seq[Filter],
- pushDownLimit: Option[Int],
+ override val pushDownLimit: Option[Int],
bucketedScanDisabled: Boolean = false)
extends PaimonBaseScan(table, requiredSchema, filters, reservedFilters, pushDownLimit)
with SupportsRuntimeFiltering
diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonSplitScan.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonSplitScan.scala
index 9f792de48..191c0104e 100644
--- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonSplitScan.scala
+++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonSplitScan.scala
@@ -19,35 +19,50 @@
package org.apache.paimon.spark
import org.apache.paimon.CoreOptions
-import org.apache.paimon.spark.schema.PaimonMetadataColumn
-import org.apache.paimon.table.Table
+import org.apache.paimon.predicate.Predicate
+import org.apache.paimon.table.{KnownSplitsTable, Table}
import org.apache.paimon.table.source.{DataSplit, Split}
import org.apache.spark.sql.connector.read.{Batch, Scan}
import org.apache.spark.sql.types.StructType
+class PaimonSplitScanBuilder(table: KnownSplitsTable) extends PaimonBaseScanBuilder(table) {
+ override def build(): Scan = {
+ PaimonSplitScan(table, table.splits(), requiredSchema, pushed.map(_._2))
+ }
+}
+
/** For internal use only. */
case class PaimonSplitScan(
table: Table,
dataSplits: Array[DataSplit],
- metadataColumns: Seq[PaimonMetadataColumn] = Seq.empty)
- extends Scan
+ requiredSchema: StructType,
+ filters: Seq[Predicate])
+ extends ColumnPruningAndPushDown
with ScanHelper {
override val coreOptions: CoreOptions = CoreOptions.fromMap(table.options())
- override def readSchema(): StructType = SparkTypeUtils.fromPaimonRowType(table.rowType())
-
override def toBatch: Batch = {
PaimonBatch(
getInputPartitions(dataSplits.asInstanceOf[Array[Split]]),
- table.newReadBuilder,
+ readBuilder,
metadataColumns)
}
+
+ override def description(): String = {
+ val pushedFiltersStr = if (filters.nonEmpty) {
+ ", PushedFilters: [" + filters.mkString(",") + "]"
+ } else {
+ ""
+ }
+ s"PaimonSplitScan: [${table.name}]" + pushedFiltersStr
+ }
}
object PaimonSplitScan {
def apply(table: Table, dataSplits: Array[DataSplit]): PaimonSplitScan = {
- new PaimonSplitScan(table, dataSplits)
+ val requiredSchema = SparkTypeUtils.fromPaimonRowType(table.rowType)
+ new PaimonSplitScan(table, dataSplits, requiredSchema, Seq.empty)
}
}
diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/SparkTable.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/SparkTable.scala
index d06c125d2..e45d15aa9 100644
--- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/SparkTable.scala
+++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/SparkTable.scala
@@ -21,7 +21,7 @@ package org.apache.paimon.spark
import org.apache.paimon.CoreOptions
import org.apache.paimon.options.Options
import org.apache.paimon.spark.schema.PaimonMetadataColumn
-import org.apache.paimon.table.{DataTable, FileStoreTable, Table}
+import org.apache.paimon.table.{DataTable, FileStoreTable, KnownSplitsTable, Table}
import org.apache.paimon.utils.StringUtils
import org.apache.spark.sql.connector.catalog.{MetadataColumn, SupportsMetadataColumns, SupportsRead, SupportsWrite, TableCapability, TableCatalog}
@@ -91,7 +91,12 @@ case class SparkTable(table: Table)
}
override def newScanBuilder(options: CaseInsensitiveStringMap): ScanBuilder = {
- new PaimonScanBuilder(table.copy(options.asCaseSensitiveMap))
+ table match {
+ case t: KnownSplitsTable =>
+ new PaimonSplitScanBuilder(t)
+ case _ =>
+ new PaimonScanBuilder(table.copy(options.asCaseSensitiveMap))
+ }
}
override def newWriteBuilder(info: LogicalWriteInfo): WriteBuilder = {
diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/MergeIntoPaimonTable.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/MergeIntoPaimonTable.scala
index f41858067..51ae6e086 100644
--- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/MergeIntoPaimonTable.scala
+++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/MergeIntoPaimonTable.scala
@@ -32,7 +32,7 @@ import org.apache.spark.sql.{Column, Dataset, Row, SparkSession}
import org.apache.spark.sql.PaimonUtils._
import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.catalyst.encoders.ExpressionEncoder
-import org.apache.spark.sql.catalyst.expressions.{Alias, Attribute, BasePredicate, EqualTo, Expression, Literal, Or, UnsafeProjection}
+import org.apache.spark.sql.catalyst.expressions.{Alias, Attribute, BasePredicate, Expression, Literal, UnsafeProjection}
import org.apache.spark.sql.catalyst.expressions.Literal.TrueLiteral
import org.apache.spark.sql.catalyst.expressions.codegen.GeneratePredicate
import org.apache.spark.sql.catalyst.plans.logical._
diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonCommand.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonCommand.scala
index e8caea3cd..62b8ee849 100644
--- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonCommand.scala
+++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonCommand.scala
@@ -22,12 +22,12 @@ import org.apache.paimon.deletionvectors.BitmapDeletionVector
import org.apache.paimon.fs.Path
import org.apache.paimon.index.IndexFileMeta
import org.apache.paimon.io.{CompactIncrement, DataFileMeta, DataIncrement, IndexIncrement}
-import org.apache.paimon.spark.{PaimonSplitScan, SparkFilterConverter}
-import org.apache.paimon.spark.catalyst.Compatibility
+import org.apache.paimon.spark.{SparkFilterConverter, SparkTable}
import org.apache.paimon.spark.catalyst.analysis.expressions.ExpressionHelper
import org.apache.paimon.spark.commands.SparkDataFileMeta.convertToSparkDataFileMeta
import org.apache.paimon.spark.schema.PaimonMetadataColumn
import org.apache.paimon.spark.schema.PaimonMetadataColumn._
+import org.apache.paimon.table.{FileStoreTable, KnownSplitsTable}
import org.apache.paimon.table.sink.{CommitMessage, CommitMessageImpl}
import org.apache.paimon.table.source.DataSplit
import org.apache.paimon.types.RowType
@@ -35,10 +35,11 @@ import org.apache.paimon.utils.SerializationUtils
import org.apache.spark.sql.{Dataset, Row, SparkSession}
import org.apache.spark.sql.PaimonUtils.createDataset
+import org.apache.spark.sql.catalyst.SQLConfHelper
import org.apache.spark.sql.catalyst.expressions.{Attribute, Expression}
import org.apache.spark.sql.catalyst.expressions.Literal.TrueLiteral
-import org.apache.spark.sql.catalyst.plans.logical.{Filter => FilterLogicalNode, LogicalPlan}
-import org.apache.spark.sql.execution.datasources.v2.{DataSourceV2Relation, DataSourceV2ScanRelation}
+import org.apache.spark.sql.catalyst.plans.logical.{Filter => FilterLogicalNode, LogicalPlan, Project}
+import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Relation
import org.apache.spark.sql.sources.{AlwaysTrue, And, EqualNullSafe, EqualTo, Filter}
import java.net.URI
@@ -47,7 +48,7 @@ import java.util.Collections
import scala.collection.JavaConverters._
/** Helper trait for all paimon commands. */
-trait PaimonCommand extends WithFileStoreTable with ExpressionHelper {
+trait PaimonCommand extends WithFileStoreTable with ExpressionHelper with SQLConfHelper {
/**
* For the 'INSERT OVERWRITE' semantics of SQL, Spark DataSourceV2 will call the `truncate`
@@ -115,8 +116,7 @@ trait PaimonCommand extends WithFileStoreTable with ExpressionHelper {
}
}
- val metadataCols = Seq(FILE_PATH)
- val filteredRelation = createNewScanPlan(candidateDataSplits, condition, relation, metadataCols)
+ val filteredRelation = createNewScanPlan(candidateDataSplits, condition, relation)
findTouchedFiles(createDataset(sparkSession, filteredRelation), sparkSession)
}
@@ -132,21 +132,58 @@ trait PaimonCommand extends WithFileStoreTable with ExpressionHelper {
.map(relativePath)
}
+ protected def createNewScanPlan(
+ candidateDataSplits: Seq[DataSplit],
+ condition: Expression,
+ relation: DataSourceV2Relation,
+ metadataColumns: Seq[PaimonMetadataColumn]): LogicalPlan = {
+ val newRelation = createNewScanPlan(candidateDataSplits, condition, relation)
+ val resolvedMetadataColumns = metadataColumns.map {
+ col =>
+ val attr = newRelation.resolve(col.name :: Nil, conf.resolver)
+ assert(attr.isDefined)
+ attr.get
+ }
+ Project(relation.output ++ resolvedMetadataColumns, newRelation)
+ }
+
+ protected def createNewScanPlan(
+ candidateDataSplits: Seq[DataSplit],
+ condition: Expression,
+ relation: DataSourceV2Relation): LogicalPlan = {
+ val newRelation = createNewRelation(candidateDataSplits, relation)
+ FilterLogicalNode(condition, newRelation)
+ }
+
protected def createNewRelation(
filePaths: Array[String],
filePathToMeta: Map[String, SparkDataFileMeta],
- relation: DataSourceV2Relation): (Array[SparkDataFileMeta], DataSourceV2ScanRelation) = {
+ relation: DataSourceV2Relation): (Array[SparkDataFileMeta], DataSourceV2Relation) = {
val files = filePaths.map(
file => filePathToMeta.getOrElse(file, throw new RuntimeException(s"Missing file: $file")))
val touchedDataSplits =
SparkDataFileMeta.convertToDataSplits(files, rawConvertible = true, fileStore.pathFactory())
- val newRelation = Compatibility.createDataSourceV2ScanRelation(
- relation,
- PaimonSplitScan(table, touchedDataSplits),
- relation.output)
+ val newRelation = createNewRelation(touchedDataSplits, relation)
(files, newRelation)
}
+ protected def createNewRelation(
+ splits: Seq[DataSplit],
+ relation: DataSourceV2Relation): DataSourceV2Relation = {
+ assert(relation.table.isInstanceOf[SparkTable])
+ val sparkTable = relation.table.asInstanceOf[SparkTable]
+ assert(sparkTable.table.isInstanceOf[FileStoreTable])
+ val knownSplitsTable =
+ KnownSplitsTable.create(sparkTable.table.asInstanceOf[FileStoreTable], splits.toArray)
+ // We re-plan the relation to skip analyze phase, so we should append all
+ // metadata columns manually and let Spark do column pruning during optimization.
+ relation.copy(
+ table = relation.table.asInstanceOf[SparkTable].copy(table = knownSplitsTable),
+ output = relation.output ++ sparkTable.metadataColumns.map(
+ _.asInstanceOf[PaimonMetadataColumn].toAttribute)
+ )
+ }
+
/** Notice that, the key is a relative path, not just the file name. */
protected def candidateFileMap(
candidateDataSplits: Seq[DataSplit]): Map[String, SparkDataFileMeta] = {
@@ -165,8 +202,7 @@ trait PaimonCommand extends WithFileStoreTable with ExpressionHelper {
condition: Expression,
relation: DataSourceV2Relation,
sparkSession: SparkSession): Dataset[SparkDeletionVectors] = {
- val metadataCols = Seq(FILE_PATH, ROW_INDEX)
- val filteredRelation = createNewScanPlan(candidateDataSplits, condition, relation, metadataCols)
+ val filteredRelation = createNewScanPlan(candidateDataSplits, condition, relation)
val dataWithMetadataColumns = createDataset(sparkSession, filteredRelation)
collectDeletionVectors(dataFilePathToMeta, dataWithMetadataColumns, sparkSession)
}
@@ -218,20 +254,6 @@ trait PaimonCommand extends WithFileStoreTable with ExpressionHelper {
}
}
- protected def createNewScanPlan(
- candidateDataSplits: Seq[DataSplit],
- condition: Expression,
- relation: DataSourceV2Relation,
- metadataCols: Seq[PaimonMetadataColumn]): LogicalPlan = {
- val metadataProj = metadataCols.map(_.toAttribute)
- val newRelation = relation.copy(output = relation.output ++ metadataProj)
- val scan = PaimonSplitScan(table, candidateDataSplits.toArray, metadataCols)
- FilterLogicalNode(
- condition,
- Compatibility.createDataSourceV2ScanRelation(newRelation, scan, newRelation.output))
-
- }
-
protected def buildDeletedCommitMessage(
deletedFiles: Array[SparkDataFileMeta]): Seq[CommitMessage] = {
deletedFiles
diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/UpdatePaimonTableCommand.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/UpdatePaimonTableCommand.scala
index dd88f388c..7a1124125 100644
--- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/UpdatePaimonTableCommand.scala
+++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/UpdatePaimonTableCommand.scala
@@ -18,8 +18,6 @@
package org.apache.paimon.spark.commands
-import org.apache.paimon.spark.PaimonSplitScan
-import org.apache.paimon.spark.catalyst.Compatibility
import org.apache.paimon.spark.catalyst.analysis.AssignmentAlignmentHelper
import org.apache.paimon.spark.leafnode.PaimonLeafRunnableCommand
import org.apache.paimon.spark.schema.SparkSystemColumns.ROW_KIND_COL
@@ -33,7 +31,7 @@ import org.apache.spark.sql.PaimonUtils.createDataset
import org.apache.spark.sql.catalyst.expressions.{Alias, Expression, If}
import org.apache.spark.sql.catalyst.expressions.Literal.TrueLiteral
import org.apache.spark.sql.catalyst.plans.logical.{Assignment, Filter, Project, SupportsSubquery}
-import org.apache.spark.sql.execution.datasources.v2.{DataSourceV2Relation, DataSourceV2ScanRelation}
+import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Relation
import org.apache.spark.sql.functions.lit
case class UpdatePaimonTableCommand(
@@ -138,10 +136,7 @@ case class UpdatePaimonTableCommand(
new Column(update).as(origin.name, origin.metadata)
}
- val toUpdateScanRelation = Compatibility.createDataSourceV2ScanRelation(
- relation,
- PaimonSplitScan(table, touchedDataSplits),
- relation.output)
+ val toUpdateScanRelation = createNewRelation(touchedDataSplits, relation)
val newPlan = if (condition == TrueLiteral) {
toUpdateScanRelation
} else {
@@ -153,7 +148,7 @@ case class UpdatePaimonTableCommand(
private def writeUpdatedAndUnchangedData(
sparkSession: SparkSession,
- toUpdateScanRelation: DataSourceV2ScanRelation): Seq[CommitMessage] = {
+ toUpdateScanRelation: DataSourceV2Relation): Seq[CommitMessage] = {
val updateColumns = updateExpressions.zip(relation.output).map {
case (update, origin) =>
val updated = if (condition == TrueLiteral) {
diff --git a/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/sql/DeletionVectorTest.scala b/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/sql/DeletionVectorTest.scala
index 3adccaa7c..e944429e4 100644
--- a/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/sql/DeletionVectorTest.scala
+++ b/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/sql/DeletionVectorTest.scala
@@ -21,20 +21,78 @@ package org.apache.paimon.spark.sql
import org.apache.paimon.data.BinaryRow
import org.apache.paimon.deletionvectors.{DeletionVector, DeletionVectorsMaintainer}
import org.apache.paimon.fs.Path
-import org.apache.paimon.spark.PaimonSparkTestBase
+import org.apache.paimon.spark.{PaimonSparkTestBase, PaimonSplitScan}
+import org.apache.paimon.spark.schema.PaimonMetadataColumn
import org.apache.paimon.table.FileStoreTable
+import org.apache.spark.paimon.Utils
import org.apache.spark.sql.Row
+import org.apache.spark.sql.execution.{QueryExecution, SparkPlan}
+import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanHelper
+import org.apache.spark.sql.execution.columnar.InMemoryTableScanExec
+import org.apache.spark.sql.execution.datasources.v2.{BatchScanExec, DataSourceV2Relation}
import org.apache.spark.sql.functions.lit
+import org.apache.spark.sql.util.QueryExecutionListener
import org.junit.jupiter.api.Assertions
import scala.collection.JavaConverters._
+import scala.collection.mutable.ArrayBuffer
import scala.util.Random
-class DeletionVectorTest extends PaimonSparkTestBase {
+class DeletionVectorTest extends PaimonSparkTestBase with AdaptiveSparkPlanHelper {
import testImplicits._
+ private def runAndCheckSplitScan(query: String): Unit = {
+ val batchScans = new ArrayBuffer[(DataSourceV2Relation, BatchScanExec)]()
+ val listener = new QueryExecutionListener {
+ override def onFailure(f: String, qe: QueryExecution, e: Exception): Unit = {}
+
+ private def isValidSplitScan(scan: BatchScanExec): Boolean = {
+ if (!scan.scan.isInstanceOf[PaimonSplitScan]) {
+ return false
+ }
+ val splitScan = scan.scan.asInstanceOf[PaimonSplitScan]
+ assert(splitScan.table.primaryKeys().isEmpty)
+ splitScan.coreOptions.deletionVectorsEnabled() &&
+ scan.output.exists(
+ attr => PaimonMetadataColumn.SUPPORTED_METADATA_COLUMNS.contains(attr.name))
+ }
+
+ private def appendScan(qe: QueryExecution, plan: SparkPlan): Unit = {
+ plan match {
+ case memory: InMemoryTableScanExec =>
+ foreach(memory.relation.cachedPlan)(p => appendScan(qe, p))
+ case scan: BatchScanExec if isValidSplitScan(scan) =>
+ val logicalScan = qe.analyzed.find(_.isInstanceOf[DataSourceV2Relation])
+ assert(logicalScan.isDefined)
+ batchScans.append((logicalScan.get.asInstanceOf[DataSourceV2Relation], scan))
+ case _ =>
+ }
+ }
+
+ override def onSuccess(funcName: String, qe: QueryExecution, duration: Long): Unit = {
+ foreach(qe.executedPlan)(p => appendScan(qe, p))
+ }
+ }
+ spark.listenerManager.register(listener)
+
+ try {
+ val df = spark.sql(query)
+ df.collect()
+ Utils.waitUntilEventEmpty(df.sparkSession)
+ assert(batchScans.nonEmpty, query)
+ assert(
+ batchScans.forall {
+ case (logicalScan, scan) =>
+ logicalScan.output.size > scan.output.size
+ },
+ batchScans)
+ } finally {
+ spark.listenerManager.unregister(listener)
+ }
+ }
+
bucketModes.foreach {
bucket =>
test(s"Paimon DeletionVector: merge into with bucket = $bucket") {
@@ -59,21 +117,21 @@ class DeletionVectorTest extends PaimonSparkTestBase {
val table = loadTable("target")
val dvMaintainerFactory =
new DeletionVectorsMaintainer.Factory(table.store().newIndexFileHandler())
- spark.sql(s"""
- |MERGE INTO target
- |USING source
- |ON target.a = source.a
- |WHEN MATCHED AND target.a = 5 THEN
- |UPDATE SET b = source.b + target.b
- |WHEN MATCHED AND source.c > 'c2' THEN
- |UPDATE SET *
- |WHEN MATCHED THEN
- |DELETE
- |WHEN NOT MATCHED AND c > 'c9' THEN
- |INSERT (a, b, c) VALUES (a, b * 1.1, c)
- |WHEN NOT MATCHED THEN
- |INSERT *
- |""".stripMargin)
+ runAndCheckSplitScan(s"""
+ |MERGE INTO target
+ |USING source
+ |ON target.a = source.a
+ |WHEN MATCHED AND target.a = 5 THEN
+ |UPDATE SET b = source.b + target.b
+ |WHEN MATCHED AND source.c > 'c2' THEN
+ |UPDATE SET *
+ |WHEN MATCHED THEN
+ |DELETE
+ |WHEN NOT MATCHED AND c > 'c9' THEN
+ |INSERT (a, b, c) VALUES (a, b * 1.1, c)
+ |WHEN NOT MATCHED THEN
+ |INSERT *
+ |""".stripMargin)
checkAnswer(
spark.sql("SELECT * FROM target ORDER BY a, b"),
@@ -116,7 +174,7 @@ class DeletionVectorTest extends PaimonSparkTestBase {
val cond1 = "id = 2"
val rowMetaInfo1 = getFilePathAndRowIndex(cond1)
- spark.sql(s"UPDATE T SET name = 'b_2' WHERE $cond1")
+ runAndCheckSplitScan(s"UPDATE T SET name = 'b_2' WHERE $cond1")
checkAnswer(
spark.sql(s"SELECT * from T ORDER BY id"),
Row(1, "a") :: Row(2, "b_2") :: Row(3, "c") :: Nil)
@@ -136,12 +194,12 @@ class DeletionVectorTest extends PaimonSparkTestBase {
Assertions.assertTrue(deletionVectors2 == deletionVectors3)
val cond2 = "id % 2 = 1"
- spark.sql(s"UPDATE T SET name = concat(name, '_2') WHERE $cond2")
+ runAndCheckSplitScan(s"UPDATE T SET name = concat(name, '_2') WHERE $cond2")
checkAnswer(
spark.sql(s"SELECT * from T ORDER BY id"),
Row(1, "a_2") :: Row(2, "b_2") :: Row(3, "c_2") :: Row(4, "d") :: Row(5, "e_2") :: Nil)
- spark.sql(s"UPDATE T SET name = '_all'")
+ runAndCheckSplitScan("UPDATE T SET name = '_all'")
checkAnswer(
spark.sql(s"SELECT * from T ORDER BY id"),
Row(1, "_all") :: Row(2, "_all") :: Row(3, "_all") :: Row(4, "_all") :: Row(
@@ -188,7 +246,7 @@ class DeletionVectorTest extends PaimonSparkTestBase {
val cond1 = "id = 2"
val rowMetaInfo1 = getFilePathAndRowIndex(cond1)
- spark.sql(s"UPDATE T SET name = 'b_2' WHERE $cond1")
+ runAndCheckSplitScan(s"UPDATE T SET name = 'b_2' WHERE $cond1")
checkAnswer(
spark.sql(s"SELECT * from T ORDER BY id"),
Row(1, "a", "2024") :: Row(2, "b_2", "2024") :: Row(3, "c", "2025") :: Row(
@@ -209,7 +267,7 @@ class DeletionVectorTest extends PaimonSparkTestBase {
val cond2 = "pt = '2025'"
val rowMetaInfo2 = rowMetaInfo1 ++ getFilePathAndRowIndex(cond2)
- spark.sql(s"UPDATE T SET name = concat(name, '_2') WHERE $cond2")
+ runAndCheckSplitScan(s"UPDATE T SET name = concat(name, '_2') WHERE $cond2")
checkAnswer(
spark.sql(s"SELECT * from T ORDER BY id"),
Row(1, "a", "2024") :: Row(2, "b_2", "2024") :: Row(3, "c_2", "2025") :: Row(
@@ -273,7 +331,7 @@ class DeletionVectorTest extends PaimonSparkTestBase {
val cond1 = "id = 2"
val rowMetaInfo1 = getFilePathAndRowIndex(cond1)
- spark.sql(s"DELETE FROM T WHERE $cond1")
+ runAndCheckSplitScan(s"DELETE FROM T WHERE $cond1")
checkAnswer(spark.sql(s"SELECT * from T ORDER BY id"), Row(1, "a") :: Nil)
val deletionVectors2 = getAllLatestDeletionVectors(table, dvMaintainerFactory)
Assertions.assertEquals(1, deletionVectors2.size)
@@ -291,7 +349,7 @@ class DeletionVectorTest extends PaimonSparkTestBase {
Assertions.assertTrue(deletionVectors2 == deletionVectors3)
val cond2 = "id % 2 = 1"
- spark.sql(s"DELETE FROM T WHERE $cond2")
+ runAndCheckSplitScan(s"DELETE FROM T WHERE $cond2")
checkAnswer(spark.sql(s"SELECT * from T ORDER BY id"), Row(2, "bb") :: Row(4, "d") :: Nil)
spark.sql("CALL sys.compact('T')")
@@ -337,7 +395,7 @@ class DeletionVectorTest extends PaimonSparkTestBase {
val cond1 = "id = 2"
val rowMetaInfo1 = getFilePathAndRowIndex(cond1)
- spark.sql(s"DELETE FROM T WHERE $cond1")
+ runAndCheckSplitScan(s"DELETE FROM T WHERE $cond1")
checkAnswer(
spark.sql(s"SELECT * from T ORDER BY id"),
Row(1, "a", "2024") :: Row(3, "c", "2025") :: Row(4, "d", "2025") :: Nil)
@@ -351,7 +409,7 @@ class DeletionVectorTest extends PaimonSparkTestBase {
val cond2 = "id = 3"
val rowMetaInfo2 = rowMetaInfo1 ++ getFilePathAndRowIndex(cond2)
- spark.sql(s"DELETE FROM T WHERE $cond2")
+ runAndCheckSplitScan(s"DELETE FROM T WHERE $cond2")
checkAnswer(
spark.sql(s"SELECT * from T ORDER BY id"),
Row(1, "a", "2024") :: Row(4, "d", "2025") :: Nil)
diff --git a/paimon-spark/paimon-spark-common/src/test/scala/org/apache/spark/paimon/Utils.scala b/paimon-spark/paimon-spark-common/src/test/scala/org/apache/spark/paimon/Utils.scala
index 1a899f500..5ea2dd861 100644
--- a/paimon-spark/paimon-spark-common/src/test/scala/org/apache/spark/paimon/Utils.scala
+++ b/paimon-spark/paimon-spark-common/src/test/scala/org/apache/spark/paimon/Utils.scala
@@ -18,6 +18,7 @@
package org.apache.spark.paimon
+import org.apache.spark.sql.SparkSession
import org.apache.spark.util.{Utils => SparkUtils}
import java.io.File
@@ -29,4 +30,7 @@ object Utils {
def createTempDir: File = SparkUtils.createTempDir()
+ def waitUntilEventEmpty(spark: SparkSession): Unit = {
+ spark.sparkContext.listenerBus.waitUntilEmpty()
+ }
}