This is an automated email from the ASF dual-hosted git repository. taiyang-li pushed a commit to branch fake_add_bolt_backend in repository https://gitbox.apache.org/repos/asf/gluten.git
commit 33107d98a199c5932c6b2a20295ffde959c9756d Author: liyang.127 <[email protected]> AuthorDate: Mon Jun 29 19:05:15 2026 +0800 [GLUTEN][CORE] Add common Paimon scan hooks and metadata plumbing Move Paimon scan validation and filter policy behind backend hooks while sharing the split metadata and scan transformer plumbing needed by native backends. --- .../substrait/rel/PaimonLocalFilesBuilder.java | 18 +- .../gluten/substrait/rel/PaimonLocalFilesNode.java | 54 +++- .../gluten/execution/OffloadPaimonScan.scala | 4 +- .../gluten/execution/PaimonScanTransformer.scala | 303 ++++++++++++++------- ...floadPaimonScan.scala => PaimonSparkShim.scala} | 23 +- .../gluten/execution/PaimonSparkShimImpl.scala | 46 ++++ .../apache/paimon/spark/source/PaimonConfig.scala | 57 ++++ .../substrait/proto/substrait/algebra.proto | 15 + .../gluten/backendsapi/SparkPlanExecApi.scala | 21 ++ .../execution/BasicScanExecTransformer.scala | 5 +- 10 files changed, 438 insertions(+), 108 deletions(-) diff --git a/gluten-paimon/src-paimon/main/java/org/apache/gluten/substrait/rel/PaimonLocalFilesBuilder.java b/gluten-paimon/src-paimon/main/java/org/apache/gluten/substrait/rel/PaimonLocalFilesBuilder.java index 4d6c33baee..d7b69eebc6 100644 --- a/gluten-paimon/src-paimon/main/java/org/apache/gluten/substrait/rel/PaimonLocalFilesBuilder.java +++ b/gluten-paimon/src-paimon/main/java/org/apache/gluten/substrait/rel/PaimonLocalFilesBuilder.java @@ -28,7 +28,14 @@ public class PaimonLocalFilesBuilder { List<Map<String, String>> partitionColumns, LocalFilesNode.ReadFileFormat fileFormat, List<String> preferredLocations, - Map<String, String> properties) { + Map<String, String> properties, + List<Integer> buckets, + List<Long> firstRowIds, + List<Long> maxSequenceNumbers, + List<Integer> splitGroups, + boolean useHiveSplit, + List<String> primaryKeys, + boolean allRawConvertible) { return new PaimonLocalFilesNode( index, paths, @@ -37,6 +44,13 @@ public class PaimonLocalFilesBuilder { partitionColumns, fileFormat, preferredLocations, - properties); + properties, + buckets, + firstRowIds, + maxSequenceNumbers, + splitGroups, + useHiveSplit, + primaryKeys, + allRawConvertible); } } diff --git a/gluten-paimon/src-paimon/main/java/org/apache/gluten/substrait/rel/PaimonLocalFilesNode.java b/gluten-paimon/src-paimon/main/java/org/apache/gluten/substrait/rel/PaimonLocalFilesNode.java index 30f47d87b3..d78a6c224d 100644 --- a/gluten-paimon/src-paimon/main/java/org/apache/gluten/substrait/rel/PaimonLocalFilesNode.java +++ b/gluten-paimon/src-paimon/main/java/org/apache/gluten/substrait/rel/PaimonLocalFilesNode.java @@ -16,11 +16,23 @@ */ package org.apache.gluten.substrait.rel; +import io.substrait.proto.ReadRel; +import io.substrait.proto.ReadRel.LocalFiles.FileOrFiles.OrcReadOptions; +import io.substrait.proto.ReadRel.LocalFiles.FileOrFiles.PaimonReadOptions; +import io.substrait.proto.ReadRel.LocalFiles.FileOrFiles.ParquetReadOptions; + import java.util.ArrayList; import java.util.List; import java.util.Map; public class PaimonLocalFilesNode extends LocalFilesNode { + private final List<Integer> buckets; + private final List<Long> firstRowIds; + private final List<Long> maxSequenceNumbers; + private final List<Integer> splitGroups; + private final boolean useHiveSplit; + private final List<String> primaryKeys; + private final boolean allRawConvertible; public PaimonLocalFilesNode( Integer index, @@ -30,7 +42,14 @@ public class PaimonLocalFilesNode extends LocalFilesNode { List<Map<String, String>> partitionColumns, ReadFileFormat fileFormat, List<String> preferredLocations, - Map<String, String> properties) { + Map<String, String> properties, + List<Integer> buckets, + List<Long> firstRowIds, + List<Long> maxSequenceNumbers, + List<Integer> splitGroups, + boolean useHiveSplit, + List<String> primaryKeys, + boolean allRawConvertible) { super( index, paths, @@ -44,5 +63,38 @@ public class PaimonLocalFilesNode extends LocalFilesNode { preferredLocations, properties, new ArrayList<>()); + this.buckets = buckets; + this.firstRowIds = firstRowIds; + this.maxSequenceNumbers = maxSequenceNumbers; + this.splitGroups = splitGroups; + this.useHiveSplit = useHiveSplit; + this.primaryKeys = primaryKeys; + this.allRawConvertible = allRawConvertible; + } + + @Override + protected void processFileBuilder(ReadRel.LocalFiles.FileOrFiles.Builder fileBuilder, int index) { + PaimonReadOptions.Builder paimonBuilder = PaimonReadOptions.newBuilder(); + paimonBuilder.setBucket(buckets.get(index)); + paimonBuilder.setFirstRowId(firstRowIds.get(index)); + paimonBuilder.setMaxSequenceNumber(maxSequenceNumbers.get(index)); + paimonBuilder.setSplitGroup(splitGroups.get(index)); + paimonBuilder.setUseHiveSplit(useHiveSplit); + paimonBuilder.addAllPrimaryKeys(primaryKeys); + paimonBuilder.setRawConvertible(allRawConvertible); + + switch (fileFormat) { + case ParquetReadFormat: + paimonBuilder.setParquet(ParquetReadOptions.newBuilder().build()); + break; + case OrcReadFormat: + paimonBuilder.setOrc(OrcReadOptions.newBuilder().build()); + break; + default: + throw new UnsupportedOperationException( + "Unsupported file format " + fileFormat.name() + " for paimon data file."); + } + + fileBuilder.setPaimon(paimonBuilder); } } diff --git a/gluten-paimon/src-paimon/main/scala/org/apache/gluten/execution/OffloadPaimonScan.scala b/gluten-paimon/src-paimon/main/scala/org/apache/gluten/execution/OffloadPaimonScan.scala index 19c0c22433..18f2151e6e 100644 --- a/gluten-paimon/src-paimon/main/scala/org/apache/gluten/execution/OffloadPaimonScan.scala +++ b/gluten-paimon/src-paimon/main/scala/org/apache/gluten/execution/OffloadPaimonScan.scala @@ -23,8 +23,8 @@ import org.apache.spark.sql.execution.datasources.v2.BatchScanExec case class OffloadPaimonScan() extends OffloadSingleNode { override def offload(plan: SparkPlan): SparkPlan = plan match { - case scan: BatchScanExec if PaimonScanTransformer.supportsBatchScan(scan.scan) => - PaimonScanTransformer(scan) + case scan: BatchScanExec if AbstractPaimonScanTransformer.supportsBatchScan(scan.scan).ok() => + AbstractPaimonScanTransformer(scan) case other => other } } diff --git a/gluten-paimon/src-paimon/main/scala/org/apache/gluten/execution/PaimonScanTransformer.scala b/gluten-paimon/src-paimon/main/scala/org/apache/gluten/execution/PaimonScanTransformer.scala index 12bccb1568..927773af78 100644 --- a/gluten-paimon/src-paimon/main/scala/org/apache/gluten/execution/PaimonScanTransformer.scala +++ b/gluten-paimon/src-paimon/main/scala/org/apache/gluten/execution/PaimonScanTransformer.scala @@ -16,37 +16,44 @@ */ package org.apache.gluten.execution +import org.apache.gluten.backendsapi.BackendsApiManager import org.apache.gluten.exception.GlutenNotSupportException import org.apache.gluten.sql.shims.SparkShimLoader import org.apache.gluten.substrait.rel.{PaimonLocalFilesBuilder, SplitInfo} import org.apache.gluten.substrait.rel.LocalFilesNode.ReadFileFormat import org.apache.spark.Partition +import org.apache.spark.internal.Logging import org.apache.spark.rdd.RDD import org.apache.spark.softaffinity.SoftAffinity import org.apache.spark.sql.catalyst.InternalRow -import org.apache.spark.sql.catalyst.catalog.ExternalCatalogUtils import org.apache.spark.sql.catalyst.expressions.{AttributeReference, DynamicPruningExpression, Expression, Literal} import org.apache.spark.sql.catalyst.plans.QueryPlan +import org.apache.spark.sql.catalyst.util.CharVarcharUtils import org.apache.spark.sql.connector.catalog.Table import org.apache.spark.sql.connector.read.Scan import org.apache.spark.sql.execution.datasources.v2.BatchScanExec +import org.apache.spark.sql.internal.SQLConf import org.apache.spark.sql.types.StructType import org.apache.spark.sql.vectorized.ColumnarBatch import org.apache.paimon.CoreOptions -import org.apache.paimon.spark.{PaimonInputPartition, PaimonScan} +import org.apache.paimon.spark.{PaimonBaseScan, PaimonInputPartition, PaimonScan} +import org.apache.paimon.spark.schema.PaimonMetadataColumn.SUPPORTED_METADATA_COLUMNS +import org.apache.paimon.spark.source.PaimonConfig import org.apache.paimon.table.{DataTable, FileStoreTable} import org.apache.paimon.table.source.DataSplit -import org.apache.paimon.utils.InternalRowPartitionComputer +import org.apache.paimon.types.DecimalType +import java.lang.{Integer => JInteger} import java.lang.{Long => JLong} import java.util.{HashMap => JHashMap, Map => JMap} import scala.collection.JavaConverters._ +import scala.collection.immutable.HashMap import scala.collection.mutable -case class PaimonScanTransformer( +abstract class AbstractPaimonScanTransformer( override val output: Seq[AttributeReference], @transient override val scan: Scan, override val runtimeFilters: Seq[Expression], @@ -58,14 +65,36 @@ case class PaimonScanTransformer( output = output, scan = scan, runtimeFilters = runtimeFilters, - table = table, keyGroupedPartitioning = keyGroupedPartitioning, + table = table, commonPartitionValues = commonPartitionValues - ) { + ) + with Logging { + + @transient private lazy val shim: PaimonSparkShim = new PaimonSparkShimImpl() + + protected lazy val tableProperties: HashMap[String, String] = { + val map = HashMap.newBuilder[String, String] + scan match { + case paimonScan: PaimonScan => + val paimonTable = paimonScan.table + val coreOptions = new CoreOptions(paimonTable.options()) + coreOptions.toMap.forEach((key, value) => map += (key -> value)) + if (SQLConf.get.getConf(PaimonConfig.PAIMON_NATIVE_SPLIT_ENABLED)) { + map += ("isPaimon" -> "true") + } + val primaryKeys = paimonTable.primaryKeys() + if (!primaryKeys.isEmpty) { + map += ("primary-key" -> String.join(",", primaryKeys)) + } + case _ => + } + map.result() + } private lazy val coreOptions: CoreOptions = scan match { - case scan: PaimonScan => - scan.table match { + case paimonScan: PaimonScan => + paimonScan.table match { case dataTable: DataTable => dataTable.coreOptions() case _ => @@ -83,15 +112,8 @@ case class PaimonScanTransformer( throw new GlutenNotSupportException("Only support PaimonScan.") } - override def getDataSchema: StructType = new StructType() - - override def withNewPushdownFilters(filters: Seq[Expression]): PaimonScanTransformer = { - this.copy(pushDownFilters = Some(filters)) - } - - override def withOutput(newOutput: Seq[AttributeReference]): BatchScanExecTransformerBase = { - this.copy(output = newOutput) - } + override def getDataSchema: StructType = CharVarcharUtils.replaceCharVarcharWithStringInSchema( + scan.asInstanceOf[PaimonBaseScan].readSchema()) override lazy val fileFormat: ReadFileFormat = { val formatStr = coreOptions.fileFormatString() @@ -105,100 +127,113 @@ case class PaimonScanTransformer( } override def doValidateInternal(): ValidationResult = { - scan match { - case paimonScan: PaimonScan => - paimonScan.table match { - case table: FileStoreTable => - if (fileFormat == ReadFileFormat.UnknownFormat) { - return ValidationResult.failed("Only support parquet/orc Paimon table.") - } - if (!table.primaryKeys().isEmpty || coreOptions.deletionVectorsEnabled()) { - return ValidationResult.failed("Not support Paimon PK/DV table.") - } - case table => - return ValidationResult.failed( - s"Not support Paimon ${table.getClass.getSimpleName} table.") - } - case _ => - return ValidationResult.failed("Only support PaimonScan.") + val result = AbstractPaimonScanTransformer.supportsBatchScan(scan) + if (result.ok()) { + return super.doValidateInternal() } - super.doValidateInternal() + result } override def doExecuteColumnar(): RDD[ColumnarBatch] = throw new UnsupportedOperationException() + override protected def rewriteNativeScanFilters(filters: Seq[Expression]): Seq[Expression] = + scan match { + case paimonScan: PaimonScan => + BackendsApiManager.getSparkPlanExecApiInstance.rewritePaimonPushdownFilters( + filters, + paimonScan.table.primaryKeys().asScala.toSet, + SUPPORTED_METADATA_COLUMNS.toSet) + case _ => filters + } + override def getSplitInfosFromPartitions( partitions: Seq[(Partition, ReadFileFormat)]): Seq[SplitInfo] = { - val partitionComputer = PaimonScanTransformer.getRowDataPartitionComputer(scan) - partitions.map { case (partition, _) => partitionToSplitInfo(partition, partitionComputer) } - } + val partitionComputer = scan match { + case paimonScan: PaimonScan => shim.getInternalPartitionComputer(paimonScan) + case _ => throw new GlutenNotSupportException("Only support PaimonScan.") + } + val paimonScan = scan.asInstanceOf[PaimonScan] + val primaryKeys = paimonScan.table.primaryKeys() + val useHiveSplit = !SQLConf.get.getConf(PaimonConfig.PAIMON_NATIVE_SPLIT_ENABLED) - private def partitionToSplitInfo( - partition: Partition, - partitionComputer: InternalRowPartitionComputer): SplitInfo = { - partition match { - case p: SparkDataSourceRDDPartition => + partitions.map { + case (partition: SparkDataSourceRDDPartition, _) => val paths = mutable.ListBuffer.empty[String] val starts = mutable.ListBuffer.empty[JLong] val lengths = mutable.ListBuffer.empty[JLong] val partitionColumns = mutable.ListBuffer.empty[JMap[String, String]] + val buckets = mutable.ListBuffer.empty[JInteger] + val firstRowIds = mutable.ListBuffer.empty[JLong] + val maxSequenceNumbers = mutable.ListBuffer.empty[JLong] + val splitGroups = mutable.ListBuffer.empty[JInteger] + val allRawConvertible = partition.inputPartitions.forall { + case paimonPartition: PaimonInputPartition => + paimonPartition.splits.forall(_.convertToRawFiles().isPresent) + case _ => false + } - p.inputPartitions.foreach { - case partition: PaimonInputPartition => - partition.splits.foreach { - split => - val rawFilesOpt = split.convertToRawFiles() - if (rawFilesOpt.isPresent) { - val partitionRow = split.asInstanceOf[DataSplit].partition() - val partitionCols = partitionComputer.generatePartValues(partitionRow) - val rawFiles = rawFilesOpt.get().asScala - paths ++= rawFiles.map(_.path()) - starts ++= rawFiles.map(file => JLong.valueOf(file.offset())) - lengths ++= rawFiles.map(file => JLong.valueOf(file.length())) - partitionColumns ++= mutable.ArrayBuffer.fill(rawFiles.size)(partitionCols) - } else { - throw new GlutenNotSupportException( - "Cannot get raw files from paimon SparkInputPartition.") - } + partition.inputPartitions.foreach { + case paimonPartition: PaimonInputPartition => + paimonPartition.splits.zipWithIndex.foreach { + case (split: DataSplit, splitIdx) => + val partitionRow = + partitionComputer.generatePartValues(shim.getSplitPartition(split)) + val fileMetas = split.dataFiles().asScala + val bucket = split.bucket() + paths ++= fileMetas.map( + file => { + val bucketPath = shim.getBucketPath(split, file) + bucketPath + "/" + file.fileName() + }) + starts ++= mutable.ArrayBuffer.fill(fileMetas.size)(JLong.valueOf(0L)) + lengths ++= fileMetas.map(file => JLong.valueOf(file.fileSize())) + partitionColumns ++= mutable.ArrayBuffer.fill(fileMetas.size)(partitionRow) + buckets ++= fileMetas.map(_ => JInteger.valueOf(bucket)) + firstRowIds ++= fileMetas + .map(_.firstRowId()) + .map(id => JLong.valueOf(if (id == null) 0L else id.toLong)) + maxSequenceNumbers ++= fileMetas.map( + file => JLong.valueOf(file.maxSequenceNumber())) + splitGroups ++= fileMetas.map(_ => JInteger.valueOf(splitIdx)) + case (other, _) => + throw new UnsupportedOperationException( + s"paimon split type: '${other.getClass.getName}' is not supported") } - case o => - throw new GlutenNotSupportException(s"Unsupported input partition type: $o") + case other => + throw new GlutenNotSupportException(s"Unsupported input partition type: $other") } + val preferredLoc = + SoftAffinity.getFilePartitionLocations(paths.toArray, partition.preferredLocations()) PaimonLocalFilesBuilder.makePaimonLocalFiles( - p.index, + partition.index, paths.asJava, starts.asJava, lengths.asJava, partitionColumns.asJava, fileFormat, - SoftAffinity - .getFilePartitionLocations(paths.toArray, p.preferredLocations()) - .toList - .asJava, - new JHashMap[String, String]() + preferredLoc.toList.asJava, + new JHashMap[String, String](), + buckets.asJava, + firstRowIds.asJava, + maxSequenceNumbers.asJava, + splitGroups.asJava, + useHiveSplit, + primaryKeys, + allRawConvertible ) - case _ => throw new GlutenNotSupportException() + case (other, _) => + throw new GlutenNotSupportException(s"Unsupported input partition type: $other") } } - override def doCanonicalize(): PaimonScanTransformer = { - this.copy( - output = output.map(QueryPlan.normalizeExpressions(_, output)), - runtimeFilters = QueryPlan.normalizePredicates( - runtimeFilters.filterNot(_ == DynamicPruningExpression(Literal.TrueLiteral)), - output), - pushDownFilters = pushDownFilters.map(QueryPlan.normalizePredicates(_, output)) - ) - } - override protected[this] def supportsBatchScan(scan: Scan): Boolean = - PaimonScanTransformer.supportsBatchScan(scan) + AbstractPaimonScanTransformer.supportsBatchScan(scan).ok() } -object PaimonScanTransformer { +object AbstractPaimonScanTransformer { def apply(batchScan: BatchScanExec): PaimonScanTransformer = { - new PaimonScanTransformer( + PaimonScanTransformer( batchScan.output, batchScan.scan, batchScan.runtimeFilters, @@ -208,23 +243,105 @@ object PaimonScanTransformer { ) } - private def getRowDataPartitionComputer(scan: Scan): InternalRowPartitionComputer = { + def supportsBatchScan(scan: Scan): ValidationResult = { + if (!SQLConf.get.getConf(PaimonConfig.PAIMON_NATIVE_SOURCE_ENABLED)) { + return ValidationResult.failed("[Paimon Fallback]: The paimon native source is not enabled.") + } + scan match { case paimonScan: PaimonScan => - val table = paimonScan.table.asInstanceOf[FileStoreTable] - // use __HIVE_DEFAULT_PARTITION__ because velox using this - new InternalRowPartitionComputer( - ExternalCatalogUtils.DEFAULT_PARTITION_NAME, - table.schema().logicalPartitionType(), - table.partitionKeys.asScala.toArray, - false + val table = paimonScan.table + val coreOptions = new CoreOptions(table.options()) + val partitionKeys = table.partitionKeys() + val partitionType = table.rowType().project(partitionKeys) + + if (!table.isInstanceOf[FileStoreTable]) { + return ValidationResult.failed( + s"[Paimon Fallback]: The table is not fileStoreTable: ${table.getClass}") + } + + if (partitionType.getFieldTypes.toArray.exists(_.isInstanceOf[DecimalType])) { + return ValidationResult.failed( + "[Paimon Fallback]: Not support decimal type as partition column") + } + + val formatString = coreOptions.fileFormatString() + if ( + !formatString.equalsIgnoreCase(CoreOptions.FILE_FORMAT_PARQUET) && + !formatString.equalsIgnoreCase(CoreOptions.FILE_FORMAT_ORC) + ) { + return ValidationResult.failed( + "[Paimon Fallback]: Only support parquet/orc Paimon table.") + } + + val isAllParquet = table + .asInstanceOf[FileStoreTable] + .coreOptions() + .fileFormatPerLevel() + .values() + .stream() + .allMatch(fmt => fmt.equalsIgnoreCase(CoreOptions.FILE_FORMAT_PARQUET)) || + paimonScan.coreOptions.fileFormatString().equalsIgnoreCase( + CoreOptions.FILE_FORMAT_PARQUET) + val schemaCols = scan.readSchema().fields.map(_.name).toSet + val schemaHasMetadataCols = SUPPORTED_METADATA_COLUMNS.exists(schemaCols.contains) + if (schemaHasMetadataCols && !isAllParquet) { + return ValidationResult.failed( + "[Paimon Fallback]: Metadata column queries are only supported with parquet files.") + } + + val allSplitsRawConvertible = paimonScan.toBatch.planInputPartitions().forall { + case partition: PaimonInputPartition => + partition.splits.forall(_.convertToRawFiles().isPresent) + case _ => false + } + BackendsApiManager.getSparkPlanExecApiInstance.validatePaimonScanCapabilities( + hasPrimaryKeys = !table.primaryKeys().isEmpty, + allSplitsRawConvertible = allSplitsRawConvertible, + deletionVectorsEnabled = coreOptions.deletionVectorsEnabled(), + changelogProducer = coreOptions.changelogProducer().name(), + mergeEngine = coreOptions.mergeEngine().name() ) - case _ => - throw new GlutenNotSupportException("Only support PaimonScan.") + case other => + ValidationResult.failed( + s"[Paimon Fallback] Scan is not a PaimonScan. Got ${other.getClass}") } } +} - def supportsBatchScan(scan: Scan): Boolean = { - scan.getClass == classOf[PaimonScan] +case class PaimonScanTransformer( + override val output: Seq[AttributeReference], + @transient override val scan: Scan, + override val runtimeFilters: Seq[Expression], + @transient override val table: Table, + override val keyGroupedPartitioning: Option[Seq[Expression]] = None, + override val commonPartitionValues: Option[Seq[(InternalRow, Int)]] = None, + override val pushDownFilters: Option[Seq[Expression]] = None) + extends AbstractPaimonScanTransformer( + output = output, + scan = scan, + runtimeFilters = runtimeFilters, + table = table, + keyGroupedPartitioning = keyGroupedPartitioning, + commonPartitionValues = commonPartitionValues, + pushDownFilters = pushDownFilters + ) { + + override def withNewPushdownFilters(filters: Seq[Expression]): PaimonScanTransformer = { + copy(pushDownFilters = Some(filters)) + } + + override def withOutput(newOutput: Seq[AttributeReference]): BatchScanExecTransformerBase = { + copy(output = newOutput) + } + + override def doCanonicalize(): PaimonScanTransformer = { + copy( + output = output.map(QueryPlan.normalizeExpressions(_, output)), + runtimeFilters = QueryPlan.normalizePredicates( + runtimeFilters.filterNot(_ == DynamicPruningExpression(Literal.TrueLiteral)), + output), + pushDownFilters = pushDownFilters.map(QueryPlan.normalizePredicates(_, output)) + ) } } diff --git a/gluten-paimon/src-paimon/main/scala/org/apache/gluten/execution/OffloadPaimonScan.scala b/gluten-paimon/src-paimon/main/scala/org/apache/gluten/execution/PaimonSparkShim.scala similarity index 60% copy from gluten-paimon/src-paimon/main/scala/org/apache/gluten/execution/OffloadPaimonScan.scala copy to gluten-paimon/src-paimon/main/scala/org/apache/gluten/execution/PaimonSparkShim.scala index 19c0c22433..135cebe180 100644 --- a/gluten-paimon/src-paimon/main/scala/org/apache/gluten/execution/OffloadPaimonScan.scala +++ b/gluten-paimon/src-paimon/main/scala/org/apache/gluten/execution/PaimonSparkShim.scala @@ -16,15 +16,20 @@ */ package org.apache.gluten.execution -import org.apache.gluten.extension.columnar.offload.OffloadSingleNode +import org.apache.spark.internal.Logging -import org.apache.spark.sql.execution.SparkPlan -import org.apache.spark.sql.execution.datasources.v2.BatchScanExec +import org.apache.paimon.io.DataFileMeta +import org.apache.paimon.spark.PaimonScan +import org.apache.paimon.table.source.DataSplit +import org.apache.paimon.utils.InternalRowPartitionComputer -case class OffloadPaimonScan() extends OffloadSingleNode { - override def offload(plan: SparkPlan): SparkPlan = plan match { - case scan: BatchScanExec if PaimonScanTransformer.supportsBatchScan(scan.scan) => - PaimonScanTransformer(scan) - case other => other - } +trait PaimonSparkShim extends Logging { + + def isChainSplit(split: DataSplit): Boolean + + def getSplitPartition(split: DataSplit): org.apache.paimon.data.InternalRow + + def getBucketPath(split: DataSplit, file: DataFileMeta): String + + def getInternalPartitionComputer(scan: PaimonScan): InternalRowPartitionComputer } diff --git a/gluten-paimon/src-paimon/main/scala/org/apache/gluten/execution/PaimonSparkShimImpl.scala b/gluten-paimon/src-paimon/main/scala/org/apache/gluten/execution/PaimonSparkShimImpl.scala new file mode 100644 index 0000000000..fdcc009544 --- /dev/null +++ b/gluten-paimon/src-paimon/main/scala/org/apache/gluten/execution/PaimonSparkShimImpl.scala @@ -0,0 +1,46 @@ +/* + * 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.gluten.execution + +import org.apache.spark.sql.catalyst.catalog.ExternalCatalogUtils + +import org.apache.paimon.data.InternalRow +import org.apache.paimon.io.DataFileMeta +import org.apache.paimon.spark.PaimonScan +import org.apache.paimon.table.FileStoreTable +import org.apache.paimon.table.source.DataSplit +import org.apache.paimon.utils.InternalRowPartitionComputer + +import scala.collection.JavaConverters._ + +class PaimonSparkShimImpl extends PaimonSparkShim { + + override def isChainSplit(split: DataSplit): Boolean = false + + override def getSplitPartition(split: DataSplit): InternalRow = split.partition() + + override def getBucketPath(split: DataSplit, file: DataFileMeta): String = split.bucketPath() + + override def getInternalPartitionComputer(scan: PaimonScan): InternalRowPartitionComputer = { + val table = scan.table.asInstanceOf[FileStoreTable] + new InternalRowPartitionComputer( + ExternalCatalogUtils.DEFAULT_PARTITION_NAME, + table.schema().logicalPartitionType(), + table.partitionKeys().asScala.toArray, + false) + } +} diff --git a/gluten-paimon/src-paimon/main/scala/org/apache/paimon/spark/source/PaimonConfig.scala b/gluten-paimon/src-paimon/main/scala/org/apache/paimon/spark/source/PaimonConfig.scala new file mode 100644 index 0000000000..e4fcfd7d81 --- /dev/null +++ b/gluten-paimon/src-paimon/main/scala/org/apache/paimon/spark/source/PaimonConfig.scala @@ -0,0 +1,57 @@ +/* + * 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.source + +import org.apache.spark.sql.internal.SQLConf.buildConf + +object PaimonConfig { + + val PAIMON_NATIVE_SOURCE_ENABLED = + buildConf("spark.gluten.paimon.native.source.enabled") + .doc("When true, enable the paimon native source.") + .internal() + .booleanConf + .createWithDefault(true) + + val PAIMON_NATIVE_MOR_ENABLED = + buildConf("spark.gluten.paimon.native.mor.source.enabled") + .doc("When true, enables the paimon mor source.") + .internal() + .booleanConf + .createWithDefault(false) + + val PAIMON_NATIVE_MOR_AGGREGATE_ENGINE_ENABLED = + buildConf("spark.gluten.paimon.native.mor.aggregate.engine.enabled") + .doc("When true, enables the paimon mor aggregate engine.") + .internal() + .booleanConf + .createWithDefault(false) + + val PAIMON_NATIVE_MOR_PARTIAL_UPDATE_ENGINE_ENABLED = + buildConf("spark.gluten.paimon.native.mor.partial.update.engine.enabled") + .doc("When true, enables the paimon mor partial update engine.") + .internal() + .booleanConf + .createWithDefault(false) + + val PAIMON_NATIVE_SPLIT_ENABLED = + buildConf("spark.gluten.paimon.native.split.enabled") + .doc("When true, enables the paimon split creation.") + .internal() + .booleanConf + .createWithDefault(true) +} diff --git a/gluten-substrait/src/main/resources/substrait/proto/substrait/algebra.proto b/gluten-substrait/src/main/resources/substrait/proto/substrait/algebra.proto index 02c7f4cc5c..5333aab35f 100644 --- a/gluten-substrait/src/main/resources/substrait/proto/substrait/algebra.proto +++ b/gluten-substrait/src/main/resources/substrait/proto/substrait/algebra.proto @@ -197,6 +197,20 @@ message ReadRel { repeated DeleteFile delete_files = 3; } + message PaimonReadOptions { + int32 bucket = 1; + int64 first_row_id = 2; + int64 max_sequence_number = 3; + int32 split_group = 4; + oneof file_format { + ParquetReadOptions parquet = 5; + OrcReadOptions orc = 6; + } + bool use_hive_split = 7; + bool raw_convertible = 8; + repeated string primary_keys = 9; + } + message DeltaReadOptions { enum RowIndexFilterType { KEEP_ALL = 0; @@ -220,6 +234,7 @@ message ReadRel { JsonReadOptions json = 15; IcebergReadOptions iceberg = 16; DeltaReadOptions delta = 22; + PaimonReadOptions paimon = 23; } message partitionColumn { diff --git a/gluten-substrait/src/main/scala/org/apache/gluten/backendsapi/SparkPlanExecApi.scala b/gluten-substrait/src/main/scala/org/apache/gluten/backendsapi/SparkPlanExecApi.scala index 6aa2f7f974..67a042f020 100644 --- a/gluten-substrait/src/main/scala/org/apache/gluten/backendsapi/SparkPlanExecApi.scala +++ b/gluten-substrait/src/main/scala/org/apache/gluten/backendsapi/SparkPlanExecApi.scala @@ -578,6 +578,27 @@ trait SparkPlanExecApi { def supportPushDownFilterToScan(sparkExecNode: LeafExecNode): Boolean = true + def validatePaimonScanCapabilities( + hasPrimaryKeys: Boolean, + allSplitsRawConvertible: Boolean, + deletionVectorsEnabled: Boolean, + changelogProducer: String, + mergeEngine: String): ValidationResult = { + if (deletionVectorsEnabled) { + return ValidationResult.failed( + "[Paimon Fallback]: The scan with deletion vector is not supported") + } + if (hasPrimaryKeys) { + return ValidationResult.failed("[Paimon Fallback]: Not support Paimon PK table.") + } + ValidationResult.succeeded + } + + def rewritePaimonPushdownFilters( + filters: Seq[Expression], + primaryKeys: Set[String], + metadataColumns: Set[String]): Seq[Expression] = filters + /** Return whether the filter is supported in scan. */ def isSupportedScanFilter(filter: Expression, sparkExecNode: LeafExecNode): Boolean = { ExpressionConverter.canReplaceWithExpressionTransformer( diff --git a/gluten-substrait/src/main/scala/org/apache/gluten/execution/BasicScanExecTransformer.scala b/gluten-substrait/src/main/scala/org/apache/gluten/execution/BasicScanExecTransformer.scala index 6d265100d0..d830e3fc8e 100644 --- a/gluten-substrait/src/main/scala/org/apache/gluten/execution/BasicScanExecTransformer.scala +++ b/gluten-substrait/src/main/scala/org/apache/gluten/execution/BasicScanExecTransformer.scala @@ -37,9 +37,11 @@ import scala.collection.JavaConverters._ trait BasicScanExecTransformer extends LeafTransformSupport with BaseDataSource { import org.apache.spark.sql.catalyst.util._ + protected def rewriteNativeScanFilters(filters: Seq[Expression]): Seq[Expression] = filters + /** Returns the filters that can be pushed down to native file scan */ final def filterExprs(): Seq[Expression] = { - if (pushDownFilters.nonEmpty) { + val nativeFilters = if (pushDownFilters.nonEmpty) { val (_, scanFiltersNotInPushDownFilters) = scanFilters.partition(pushDownFilters.get.contains(_)) // For filters that only exists in scan, we need to check if they are supported. @@ -59,6 +61,7 @@ trait BasicScanExecTransformer extends LeafTransformSupport with BaseDataSource scanFilters.filter( BackendsApiManager.getSparkPlanExecApiInstance.isSupportedScanFilter(_, this)) } + rewriteNativeScanFilters(nativeFilters) } /** Returns the filters that already exists in scan. */ --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
