liujiayi771 commented on code in PR #13334:
URL: https://github.com/apache/gluten/pull/13334#discussion_r4371434352
##########
gluten-paimon/src-paimon/main/java/org/apache/gluten/substrait/rel/PaimonLocalFilesNode.java:
##########
@@ -44,5 +48,35 @@ public PaimonLocalFilesNode(
preferredLocations,
properties,
new ArrayList<>());
+ this.serializedSplits = null;
+ }
+
+ public PaimonLocalFilesNode(
+ Integer index, String tablePath, List<byte[]> splits, List<String>
preferredLocations) {
+ super(
+ index,
+ java.util.Collections.nCopies(splits.size(), tablePath),
Review Comment:
Nit: could we add `import java.util.Collections;` and use
`Collections.nCopies(...)` / `Collections.emptyMap()` throughout this
constructor, consistent with the existing imports and usage of `ArrayList`,
`List`, and `Map`?
##########
gluten-substrait/src/main/resources/substrait/proto/substrait/algebra.proto:
##########
@@ -262,6 +262,9 @@ message ReadRel {
bool use_hive_split = 7;
bool raw_convertible = 8;
repeated string primary_keys = 9;
+ // Native Paimon reader input. The split is DataSplit.serialize()
output.
+ string table_path = 10;
+ bytes serialized_split = 11;
Review Comment:
Could we preserve the branch and effective read-option overrides from the
JVM scan when constructing the paimon-cpp read context?
Currently, only the table path and serialized split are passed through. The
reader therefore defaults to the main branch, and query-specific option
overrides are not forwarded. Reloading the persisted table options does not
necessarily reproduce the JVM scan's read context.
I suggest grouping the SDK input into a dedicated `PaimonCppSplit` message
inside `PaimonReadOptions`:
```proto
message PaimonCppSplit {
bytes serialized_data_split = 1;
string table_path = 2;
string branch = 3;
}
optional PaimonCppSplit cpp_split = 10;
```
Read-option overrides can use the existing properties mechanism, with the
reader applying the branch and options via `WithBranch()` and `SetOptions()`.
Please also add coverage for a non-main branch and a query-level read-option
override.
##########
backends-clickhouse/src-paimon/main/scala/org/apache/gluten/execution/CHPaimonScanTransformer.scala:
##########
@@ -0,0 +1,171 @@
+/*
+ * 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.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.rdd.RDD
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions.{AttributeReference,
DynamicPruningExpression, Expression, Literal}
+import org.apache.spark.sql.catalyst.plans.QueryPlan
+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.types.StructType
+import org.apache.spark.sql.vectorized.ColumnarBatch
+
+import org.apache.paimon.CoreOptions
+import org.apache.paimon.io.DataOutputSerializer
+import org.apache.paimon.spark.{PaimonInputPartition, PaimonScan}
+import org.apache.paimon.table.{DataTable, FileStoreTable}
+import org.apache.paimon.table.source.DataSplit
+
+import scala.collection.JavaConverters._
+
+case class CHPaimonScanTransformer(
+ 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 BatchScanExecTransformerBase(
+ output = output,
+ scan = scan,
+ runtimeFilters = runtimeFilters,
+ table = table,
+ keyGroupedPartitioning = keyGroupedPartitioning,
+ commonPartitionValues = commonPartitionValues
+ ) {
+
+ private lazy val coreOptions: CoreOptions = scan match {
+ case paimonScan: PaimonScan =>
+ paimonScan.table match {
+ case dataTable: DataTable => dataTable.coreOptions()
+ case _ => throw new GlutenNotSupportException("Only support Paimon
DataTable.")
+ }
+ case _ => throw new GlutenNotSupportException("Only support PaimonScan.")
+ }
+
+ override def getPartitionSchema: StructType = scan match {
+ case paimonScan: PaimonScan =>
+ val partitionKeys = paimonScan.table.partitionKeys()
+ StructType(scan.readSchema().filter(field =>
partitionKeys.contains(field.name)))
+ case _ => throw new GlutenNotSupportException("Only support PaimonScan.")
+ }
+
+ override def getDataSchema: StructType = scan match {
+ case paimonScan: PaimonScan =>
+ val partitionKeys = paimonScan.table.partitionKeys()
+ StructType(scan.readSchema().filterNot(field =>
partitionKeys.contains(field.name)))
+ case _ => throw new GlutenNotSupportException("Only support PaimonScan.")
+ }
+
+ override def withNewPushdownFilters(filters: Seq[Expression]):
CHPaimonScanTransformer =
+ copy(pushDownFilters = Some(filters))
+
+ override def withOutput(newOutput: Seq[AttributeReference]):
BatchScanExecTransformerBase =
+ copy(output = newOutput)
+
+ override lazy val fileFormat: ReadFileFormat = {
+ val formatStr = coreOptions.fileFormatString()
+ if ("parquet".equalsIgnoreCase(formatStr)) {
+ ReadFileFormat.ParquetReadFormat
+ } else if ("orc".equalsIgnoreCase(formatStr)) {
+ ReadFileFormat.OrcReadFormat
+ } else {
+ ReadFileFormat.UnknownFormat
+ }
+ }
+
+ override def doValidateInternal(): ValidationResult = {
+ scan match {
+ case paimonScan: PaimonScan =>
+ paimonScan.table match {
+ case _: FileStoreTable =>
+ ()
Review Comment:
Could we validate the supported split types before offloading this scan?
Checking only that the table is a `FileStoreTable` is not sufficient: Paimon
1.4.1 can produce `IncrementalSplit` for an incremental scan with
`incremental-between-scan-mode=diff`. However, `getSplitInfosFromPartitions()`
below only accepts `DataSplit` and throws for other split types. That method is
called when generating the execution RDD, after plan validation, so this
becomes a query failure rather than a JVM fallback.
Please reject unsupported split types or scan modes during validation and
add a regression test verifying that an incremental diff scan falls back and
returns the expected result. Supporting incremental reads natively can remain a
follow-up.
##########
backends-clickhouse/src-paimon/main/scala/org/apache/gluten/component/CHOffloadPaimonScan.scala:
##########
@@ -0,0 +1,41 @@
+/*
+ * 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.component
+
+import org.apache.gluten.execution.{CHPaimonCountTransformer,
CHPaimonScanTransformer, OffloadPaimonScan, PaimonScanTransformer}
+
+import org.apache.spark.sql.execution.{LocalTableScanExec, SparkPlan}
+import org.apache.spark.sql.execution.datasources.v2.BatchScanExec
+import org.apache.spark.sql.execution.datasources.v2.DataSourceV2ScanRelation
+
+import org.apache.paimon.spark.read.PaimonLocalScan
+
+case object CHOffloadPaimonScan extends OffloadPaimonScan {
Review Comment:
Could `CHOffloadPaimonScan` extend `OffloadSingleNode` directly?
It completely overrides `offload()` and does not reuse the implementation in
`OffloadPaimonScan`. The rule injection already accepts `OffloadSingleNode`, so
inheriting from the concrete Paimon rule does not seem necessary. Extending the
interface directly would also let us keep the existing `OffloadPaimonScan` case
class unchanged, avoiding an unrelated change to the shared rule.
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
