You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@gluten.apache.org by ch...@apache.org on 2024/12/17 03:05:37 UTC
(incubator-gluten) branch main updated: [GLUTEN-7028][CH][Part-12] Add Local SortExec for Partition Write in one pipeline mode (#8237)
This is an automated email from the ASF dual-hosted git repository.
changchen pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/incubator-gluten.git
The following commit(s) were added to refs/heads/main by this push:
new c1d66c4199 [GLUTEN-7028][CH][Part-12] Add Local SortExec for Partition Write in one pipeline mode (#8237)
c1d66c4199 is described below
commit c1d66c4199418e57bb170aea0234375786957b44
Author: Chang chen <ch...@kyligence.io>
AuthorDate: Tue Dec 17 11:05:33 2024 +0800
[GLUTEN-7028][CH][Part-12] Add Local SortExec for Partition Write in one pipeline mode (#8237)
* [Refactor] Pass WriteFilesExecTransformer to genWriteParameters
* [Feature] Add SortExec and Remove RemoveNativeWriteFilesSortAndProject
* [Bug Fix] collect_partition_cols and Remove ApplySquashingTransform and PlanSquashingTransform
* [Bug Fix] Fix "WARN org.apache.spark.sql.execution.datasources.BasicWriteTaskStatsTracker: Expected x files, but only saw 0."
* [Refactor]set CHConf.ENABLE_ONEPIPELINE_MERGETREE_WRITE.key to true in spark 35 for GlutenClickHouseMergeTreeWriteSuite
* Fix Rebase issue
---
.../delta/ClickhouseOptimisticTransaction.scala | 85 +++---
.../sql/execution/FileDeltaColumnarWrite.scala | 3 +-
.../sql/execution/datasources/DeltaV1Writes.scala | 74 +++++
.../execution/datasources/DeltaV1WritesSuite.scala | 100 +++++++
.../datasources/v1/write_optimization.proto | 3 +
.../gluten/backendsapi/clickhouse/CHRuleApi.scala | 1 -
.../backendsapi/clickhouse/CHTransformerApi.scala | 37 ++-
.../backendsapi/clickhouse/RuntimeConfig.scala | 17 ++
.../spark/sql/execution/CHColumnarWrite.scala | 13 +-
.../GlutenClickHouseDeltaParquetWriteSuite.scala | 1 +
.../GlutenClickHouseMergeTreeWriteSuite.scala | 332 ++++++++++-----------
.../backendsapi/velox/VeloxTransformerApi.scala | 10 +-
.../Parser/RelParsers/WriteRelParser.cpp | 58 ++--
.../Parser/RelParsers/WriteRelParser.h | 5 +-
.../Storages/MergeTree/SparkMergeTreeSink.cpp | 43 ++-
.../Storages/MergeTree/SparkMergeTreeSink.h | 17 +-
.../Storages/MergeTree/SparkMergeTreeWriter.cpp | 13 -
cpp-ch/local-engine/tests/gtest_write_pipeline.cpp | 10 +-
.../tests/gtest_write_pipeline_mergetree.cpp | 39 ++-
.../tests/json/mergetree/4_one_pipeline.json | 296 ++++++++++--------
.../apache/gluten/backendsapi/TransformerApi.scala | 5 +-
.../execution/WriteFilesExecTransformer.scala | 5 +-
22 files changed, 725 insertions(+), 442 deletions(-)
diff --git a/backends-clickhouse/src-delta-32/main/scala/org/apache/spark/sql/delta/ClickhouseOptimisticTransaction.scala b/backends-clickhouse/src-delta-32/main/scala/org/apache/spark/sql/delta/ClickhouseOptimisticTransaction.scala
index 05f7fdbfa4..cd3ce79374 100644
--- a/backends-clickhouse/src-delta-32/main/scala/org/apache/spark/sql/delta/ClickhouseOptimisticTransaction.scala
+++ b/backends-clickhouse/src-delta-32/main/scala/org/apache/spark/sql/delta/ClickhouseOptimisticTransaction.scala
@@ -29,9 +29,8 @@ import org.apache.spark.sql.delta.files._
import org.apache.spark.sql.delta.hooks.AutoCompact
import org.apache.spark.sql.delta.schema.{InnerInvariantViolationException, InvariantViolationException}
import org.apache.spark.sql.delta.sources.DeltaSQLConf
-import org.apache.spark.sql.execution.{QueryExecution, SparkPlan, SQLExecution}
-import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanExec
-import org.apache.spark.sql.execution.datasources.{BasicWriteJobStatsTracker, FileFormatWriter, GlutenWriterColumnarRules, WriteFiles, WriteJobStatsTracker}
+import org.apache.spark.sql.execution.SQLExecution
+import org.apache.spark.sql.execution.datasources.{BasicWriteJobStatsTracker, DeltaV1Writes, FileFormatWriter, GlutenWriterColumnarRules, WriteJobStatsTracker}
import org.apache.spark.sql.execution.datasources.v1.MergeTreeWriterInjects
import org.apache.spark.sql.execution.datasources.v1.clickhouse.MergeTreeFileFormatWriter
import org.apache.spark.sql.execution.datasources.v2.clickhouse.ClickHouseConfig
@@ -229,31 +228,12 @@ class ClickhouseOptimisticTransaction(
val (data, partitionSchema) = performCDCPartition(inputData)
val outputPath = deltaLog.dataPath
- val fileFormat = deltaLog.fileFormat(protocol, metadata) // TODO support changing formats.
-
- // Iceberg spec requires partition columns in data files
- val writePartitionColumns = IcebergCompat.isAnyEnabled(metadata)
- // Retain only a minimal selection of Spark writer options to avoid any potential
- // compatibility issues
- val options = (writeOptions match {
- case None => Map.empty[String, String]
- case Some(writeOptions) =>
- writeOptions.options.filterKeys {
- key =>
- key.equalsIgnoreCase(DeltaOptions.MAX_RECORDS_PER_FILE) ||
- key.equalsIgnoreCase(DeltaOptions.COMPRESSION)
- }.toMap
- }) + (DeltaOptions.WRITE_PARTITION_COLUMNS -> writePartitionColumns.toString)
-
- val (normalQueryExecution, output, generatedColumnConstraints, _) =
+ val (queryExecution, output, generatedColumnConstraints, _) =
normalizeData(deltaLog, writeOptions, data)
val partitioningColumns = getPartitioningColumns(partitionSchema, output)
- val logicalPlan = normalQueryExecution.optimizedPlan
- val write =
- WriteFiles(logicalPlan, fileFormat, partitioningColumns, None, options, Map.empty)
+ val fileFormat = deltaLog.fileFormat(protocol, metadata) // TODO support changing formats.
- val queryExecution = new QueryExecution(spark, write)
val (committer, collectStats) = fileFormat.toString match {
case "MergeTree" => (getCommitter2(outputPath), false)
case _ => (getCommitter(outputPath), true)
@@ -274,20 +254,24 @@ class ClickhouseOptimisticTransaction(
SQLExecution.withNewExecutionId(queryExecution, Option("deltaTransactionalWrite")) {
val outputSpec = FileFormatWriter.OutputSpec(outputPath.toString, Map.empty, output)
- val physicalPlan = materializeAdaptiveSparkPlan(queryExecution.executedPlan)
- // convertEmptyToNullIfNeeded(queryExecution.executedPlan, partitioningColumns, constraints)
- /* val checkInvariants = DeltaInvariantCheckerExec(empty2NullPlan, constraints)
+ val empty2NullPlan =
+ convertEmptyToNullIfNeeded(queryExecution.sparkPlan, partitioningColumns, constraints)
+ // TODO: val checkInvariants = DeltaInvariantCheckerExec(empty2NullPlan, constraints)
+ val checkInvariants = empty2NullPlan
+
// No need to plan optimized write if the write command is OPTIMIZE, which aims to produce
// evenly-balanced data files already.
- val physicalPlan =
- if (
- !isOptimize &&
- shouldOptimizeWrite(writeOptions, spark.sessionState.conf)
- ) {
- DeltaOptimizedWriterExec(checkInvariants, metadata.partitionColumns, deltaLog)
- } else {
- checkInvariants
- } */
+ // TODO: val physicalPlan =
+ // if (
+ // !isOptimize &&
+ // shouldOptimizeWrite(writeOptions, spark.sessionState.conf)
+ // ) {
+ // DeltaOptimizedWriterExec(checkInvariants, metadata.partitionColumns, deltaLog)
+ // } else {
+ // checkInvariants
+ // }
+ val physicalPlan = checkInvariants
+
val statsTrackers: ListBuffer[WriteJobStatsTracker] = ListBuffer()
if (spark.conf.get(DeltaSQLConf.DELTA_HISTORY_METRICS_ENABLED)) {
@@ -298,10 +282,33 @@ class ClickhouseOptimisticTransaction(
statsTrackers.append(basicWriteJobStatsTracker)
}
+ // Iceberg spec requires partition columns in data files
+ val writePartitionColumns = IcebergCompat.isAnyEnabled(metadata)
+ // Retain only a minimal selection of Spark writer options to avoid any potential
+ // compatibility issues
+ val options = (writeOptions match {
+ case None => Map.empty[String, String]
+ case Some(writeOptions) =>
+ writeOptions.options.filterKeys {
+ key =>
+ key.equalsIgnoreCase(DeltaOptions.MAX_RECORDS_PER_FILE) ||
+ key.equalsIgnoreCase(DeltaOptions.COMPRESSION)
+ }.toMap
+ }) + (DeltaOptions.WRITE_PARTITION_COLUMNS -> writePartitionColumns.toString)
+
+ val executedPlan = DeltaV1Writes(
+ spark,
+ physicalPlan,
+ fileFormat,
+ partitioningColumns,
+ None,
+ options
+ ).executedPlan
+
try {
DeltaFileFormatWriter.write(
sparkSession = spark,
- plan = physicalPlan,
+ plan = executedPlan,
fileFormat = fileFormat,
committer = committer,
outputSpec = outputSpec,
@@ -358,8 +365,4 @@ class ClickhouseOptimisticTransaction(
resultFiles.toSeq ++ committer.changeFiles
}
- private def materializeAdaptiveSparkPlan(plan: SparkPlan): SparkPlan = plan match {
- case a: AdaptiveSparkPlanExec => a.finalPhysicalPlan
- case p: SparkPlan => p
- }
}
diff --git a/backends-clickhouse/src-delta-32/main/scala/org/apache/spark/sql/execution/FileDeltaColumnarWrite.scala b/backends-clickhouse/src-delta-32/main/scala/org/apache/spark/sql/execution/FileDeltaColumnarWrite.scala
index bf6b0c0074..df7ef7e234 100644
--- a/backends-clickhouse/src-delta-32/main/scala/org/apache/spark/sql/execution/FileDeltaColumnarWrite.scala
+++ b/backends-clickhouse/src-delta-32/main/scala/org/apache/spark/sql/execution/FileDeltaColumnarWrite.scala
@@ -137,7 +137,8 @@ case class FileDeltaColumnarWrite(
// stats.map(row => x.apply(row).getString(0)).foreach(println)
// process stats
val commitInfo = DeltaFileCommitInfo(committer)
- val basicNativeStat = NativeBasicWriteTaskStatsTracker(description, basicWriteJobStatsTracker)
+ val basicNativeStat =
+ NativeBasicWriteTaskStatsTracker(description.path, basicWriteJobStatsTracker)
val basicNativeStats = Seq(commitInfo, basicNativeStat)
NativeStatCompute(stats)(basicNativeStats, nativeDeltaStats)
diff --git a/backends-clickhouse/src-delta-32/main/scala/org/apache/spark/sql/execution/datasources/DeltaV1Writes.scala b/backends-clickhouse/src-delta-32/main/scala/org/apache/spark/sql/execution/datasources/DeltaV1Writes.scala
new file mode 100644
index 0000000000..8ae99cc0d5
--- /dev/null
+++ b/backends-clickhouse/src-delta-32/main/scala/org/apache/spark/sql/execution/datasources/DeltaV1Writes.scala
@@ -0,0 +1,74 @@
+/*
+ * 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.spark.sql.execution.datasources
+import org.apache.gluten.backendsapi.BackendsApiManager
+
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.catalyst.catalog.BucketSpec
+import org.apache.spark.sql.catalyst.catalog.CatalogTypes.TablePartitionSpec
+import org.apache.spark.sql.catalyst.expressions.{Attribute, SortOrder}
+import org.apache.spark.sql.catalyst.plans.logical.LocalRelation
+import org.apache.spark.sql.execution.{QueryExecution, SortExec, SparkPlan}
+import org.apache.spark.sql.execution.datasources.V1WritesUtils.isOrderingMatched
+
+case class DeltaV1Writes(
+ spark: SparkSession,
+ query: SparkPlan,
+ fileFormat: FileFormat,
+ partitionColumns: Seq[Attribute],
+ bucketSpec: Option[BucketSpec],
+ options: Map[String, String],
+ staticPartitions: TablePartitionSpec = Map.empty) {
+
+ require(fileFormat != null, "FileFormat is required to write files.")
+ require(BackendsApiManager.getSettings.enableNativeWriteFiles())
+
+ private lazy val requiredOrdering: Seq[SortOrder] =
+ V1WritesUtils.getSortOrder(
+ query.output,
+ partitionColumns,
+ bucketSpec,
+ options,
+ staticPartitions.size)
+
+ lazy val sortPlan: SparkPlan = {
+ val outputOrdering = query.outputOrdering
+ val orderingMatched = isOrderingMatched(requiredOrdering.map(_.child), outputOrdering)
+ if (orderingMatched) {
+ query
+ } else {
+ SortExec(requiredOrdering, global = false, query)
+ }
+ }
+
+ lazy val writePlan: SparkPlan =
+ WriteFilesExec(
+ sortPlan,
+ fileFormat = fileFormat,
+ partitionColumns = partitionColumns,
+ bucketSpec = bucketSpec,
+ options = options,
+ staticPartitions = staticPartitions)
+
+ lazy val executedPlan: SparkPlan =
+ CallTransformer(spark, writePlan).executedPlan
+}
+
+case class CallTransformer(spark: SparkSession, physicalPlan: SparkPlan)
+ extends QueryExecution(spark, LocalRelation()) {
+ override lazy val sparkPlan: SparkPlan = physicalPlan
+}
diff --git a/backends-clickhouse/src-delta-32/test/scala/org/apache/spark/sql/execution/datasources/DeltaV1WritesSuite.scala b/backends-clickhouse/src-delta-32/test/scala/org/apache/spark/sql/execution/datasources/DeltaV1WritesSuite.scala
new file mode 100644
index 0000000000..1a90148df2
--- /dev/null
+++ b/backends-clickhouse/src-delta-32/test/scala/org/apache/spark/sql/execution/datasources/DeltaV1WritesSuite.scala
@@ -0,0 +1,100 @@
+/*
+ * 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.spark.sql.execution.datasources
+
+import org.apache.gluten.GlutenConfig
+import org.apache.gluten.execution.{GlutenClickHouseWholeStageTransformerSuite, GlutenPlan, SortExecTransformer}
+import org.apache.spark.SparkConf
+import org.apache.spark.sql.execution.datasources.parquet.ParquetFileFormat
+import org.apache.spark.sql.execution.{SortExec, SparkPlan}
+
+class DeltaV1WritesSuite extends GlutenClickHouseWholeStageTransformerSuite {
+
+ import testImplicits._
+
+ override protected def sparkConf: SparkConf = {
+ super.sparkConf
+ .set(GlutenConfig.NATIVE_WRITER_ENABLED.key, "true")
+ }
+
+ override def beforeAll(): Unit = {
+ super.beforeAll()
+ (0 to 20)
+ .map(i => (i, i % 5, (i % 10).toString))
+ .toDF("i", "j", "k")
+ .write
+ .saveAsTable("t0")
+ }
+
+ override def afterAll(): Unit = {
+ sql("drop table if exists t0")
+ super.afterAll()
+ }
+
+ val format = new ParquetFileFormat
+ def getSort(child: SparkPlan): Option[SortExecTransformer] = {
+ child.collectFirst { case w: SortExecTransformer => w }
+ }
+ test("don't add sort when the required ordering is empty") {
+ val df = sql("select * from t0")
+ val plan = df.queryExecution.sparkPlan
+ val writes = DeltaV1Writes(spark, plan, format, Nil, None, Map.empty)
+ assert(writes.sortPlan === plan)
+ assert(writes.writePlan != null)
+ assert(writes.executedPlan.isInstanceOf[GlutenPlan])
+ val writeFilesOpt = V1WritesUtils.getWriteFilesOpt(writes.executedPlan)
+ assert(writeFilesOpt.isDefined)
+ val sortExec = getSort(writes.executedPlan)
+ assert(sortExec.isEmpty)
+ }
+
+ test("don't add sort when the required ordering is already satisfied") {
+ val df = sql("select * from t0")
+ def check(plan: SparkPlan): Unit = {
+ val partitionColumns = plan.output.find(_.name == "k").toSeq
+ val writes = DeltaV1Writes(spark, plan, format, partitionColumns, None, Map.empty)
+ assert(writes.sortPlan === plan)
+ assert(writes.writePlan != null)
+ assert(writes.executedPlan.isInstanceOf[GlutenPlan])
+ val writeFilesOpt = V1WritesUtils.getWriteFilesOpt(writes.executedPlan)
+ assert(writeFilesOpt.isDefined)
+ val sortExec = getSort(writes.executedPlan)
+ assert(sortExec.isDefined)
+ }
+ check(df.orderBy("k").queryExecution.sparkPlan)
+ check(df.orderBy("k", "j").queryExecution.sparkPlan)
+ }
+
+ test("add sort when the required ordering is not satisfied") {
+ val df = sql("select * from t0")
+ def check(plan: SparkPlan): Unit = {
+ val partitionColumns = plan.output.find(_.name == "k").toSeq
+ val writes = DeltaV1Writes(spark, plan, format, partitionColumns, None, Map.empty)
+ val sort = writes.sortPlan.asInstanceOf[SortExec]
+ assert(sort.child === plan)
+ assert(writes.writePlan != null)
+ assert(writes.executedPlan.isInstanceOf[GlutenPlan])
+ val writeFilesOpt = V1WritesUtils.getWriteFilesOpt(writes.executedPlan)
+ assert(writeFilesOpt.isDefined)
+ val sortExec = getSort(writes.executedPlan)
+ assert(sortExec.isDefined, s"writes.executedPlan: ${writes.executedPlan}")
+ }
+ check(df.queryExecution.sparkPlan)
+ check(df.orderBy("j", "k").queryExecution.sparkPlan)
+ }
+
+}
diff --git a/backends-clickhouse/src/main/resources/org/apache/spark/sql/execution/datasources/v1/write_optimization.proto b/backends-clickhouse/src/main/resources/org/apache/spark/sql/execution/datasources/v1/write_optimization.proto
index 89f606e4ff..fdf34f1a0a 100644
--- a/backends-clickhouse/src/main/resources/org/apache/spark/sql/execution/datasources/v1/write_optimization.proto
+++ b/backends-clickhouse/src/main/resources/org/apache/spark/sql/execution/datasources/v1/write_optimization.proto
@@ -12,6 +12,9 @@ message Write {
message Common {
string format = 1;
string job_task_attempt_id = 2; // currently used in mergetree format
+
+ // Describes the partition index in the WriteRel.table_schema.
+ repeated int32 partition_col_index = 3;
}
message ParquetWrite{}
message OrcWrite{}
diff --git a/backends-clickhouse/src/main/scala/org/apache/gluten/backendsapi/clickhouse/CHRuleApi.scala b/backends-clickhouse/src/main/scala/org/apache/gluten/backendsapi/clickhouse/CHRuleApi.scala
index 40e5353618..32961c21a2 100644
--- a/backends-clickhouse/src/main/scala/org/apache/gluten/backendsapi/clickhouse/CHRuleApi.scala
+++ b/backends-clickhouse/src/main/scala/org/apache/gluten/backendsapi/clickhouse/CHRuleApi.scala
@@ -93,7 +93,6 @@ object CHRuleApi {
// Legacy: Post-transform rules.
injector.injectPostTransform(_ => PruneNestedColumnsInHiveTableScan)
- injector.injectPostTransform(_ => RemoveNativeWriteFilesSortAndProject())
injector.injectPostTransform(c => intercept(RewriteTransformer.apply(c.session)))
injector.injectPostTransform(_ => PushDownFilterToScan)
injector.injectPostTransform(_ => PushDownInputFileExpression.PostOffload)
diff --git a/backends-clickhouse/src/main/scala/org/apache/gluten/backendsapi/clickhouse/CHTransformerApi.scala b/backends-clickhouse/src/main/scala/org/apache/gluten/backendsapi/clickhouse/CHTransformerApi.scala
index 0be8cf2c25..ef5a4eff6f 100644
--- a/backends-clickhouse/src/main/scala/org/apache/gluten/backendsapi/clickhouse/CHTransformerApi.scala
+++ b/backends-clickhouse/src/main/scala/org/apache/gluten/backendsapi/clickhouse/CHTransformerApi.scala
@@ -17,7 +17,7 @@
package org.apache.gluten.backendsapi.clickhouse
import org.apache.gluten.backendsapi.TransformerApi
-import org.apache.gluten.execution.CHHashAggregateExecTransformer
+import org.apache.gluten.execution.{CHHashAggregateExecTransformer, WriteFilesExecTransformer}
import org.apache.gluten.expression.ConverterUtils
import org.apache.gluten.substrait.expression.{BooleanLiteralNode, ExpressionBuilder, ExpressionNode}
import org.apache.gluten.utils.{CHInputPartitionsUtil, ExpressionDocUtil}
@@ -31,7 +31,7 @@ import org.apache.spark.sql.delta.catalog.ClickHouseTableV2
import org.apache.spark.sql.delta.files.TahoeFileIndex
import org.apache.spark.sql.execution.SparkPlan
import org.apache.spark.sql.execution.aggregate.HashAggregateExec
-import org.apache.spark.sql.execution.datasources.{FileFormat, HadoopFsRelation, PartitionDirectory}
+import org.apache.spark.sql.execution.datasources.{HadoopFsRelation, PartitionDirectory}
import org.apache.spark.sql.execution.datasources.orc.OrcFileFormat
import org.apache.spark.sql.execution.datasources.parquet.ParquetFileFormat
import org.apache.spark.sql.execution.datasources.v1.Write
@@ -243,24 +243,31 @@ class CHTransformerApi extends TransformerApi with Logging {
GlutenDriverEndpoint.invalidateResourceRelation(executionId)
}
- override def genWriteParameters(
- fileFormat: FileFormat,
- writeOptions: Map[String, String]): Any = {
- val fileFormatStr = fileFormat match {
+ override def genWriteParameters(writeExec: WriteFilesExecTransformer): Any = {
+ val fileFormatStr = writeExec.fileFormat match {
case register: DataSourceRegister =>
register.shortName
case _ => "UnknownFileFormat"
}
- val write = Write
+ val childOutput = writeExec.child.output
+
+ val partitionIndexes =
+ writeExec.partitionColumns.map(p => childOutput.indexWhere(_.exprId == p.exprId))
+ require(partitionIndexes.forall(_ >= 0))
+
+ val common = Write.Common
.newBuilder()
- .setCommon(
- Write.Common
- .newBuilder()
- .setFormat(fileFormatStr)
- .setJobTaskAttemptId("") // we can get job and task id at the driver side
- .build())
+ .setFormat(s"$fileFormatStr")
+ .setJobTaskAttemptId("") // we cannot get job and task id at the driver side)
+ partitionIndexes.foreach {
+ idx =>
+ require(idx >= 0)
+ common.addPartitionColIndex(idx)
+ }
+
+ val write = Write.newBuilder().setCommon(common.build())
- fileFormat match {
+ writeExec.fileFormat match {
case d: MergeTreeFileFormat =>
write.setMergetree(MergeTreeFileFormat.createWrite(d.metadata))
case _: ParquetFileFormat =>
@@ -273,5 +280,5 @@ class CHTransformerApi extends TransformerApi with Logging {
/** use Hadoop Path class to encode the file path */
override def encodeFilePathIfNeed(filePath: String): String =
- (new Path(filePath)).toUri.toASCIIString
+ new Path(filePath).toUri.toASCIIString
}
diff --git a/backends-clickhouse/src/main/scala/org/apache/gluten/backendsapi/clickhouse/RuntimeConfig.scala b/backends-clickhouse/src/main/scala/org/apache/gluten/backendsapi/clickhouse/RuntimeConfig.scala
index 12bb8d05d9..055c3b9d87 100644
--- a/backends-clickhouse/src/main/scala/org/apache/gluten/backendsapi/clickhouse/RuntimeConfig.scala
+++ b/backends-clickhouse/src/main/scala/org/apache/gluten/backendsapi/clickhouse/RuntimeConfig.scala
@@ -22,6 +22,7 @@ object RuntimeConfig {
import CHConf._
import SQLConf._
+ /** Clickhouse Configuration */
val PATH =
buildConf(runtimeConfig("path"))
.doc(
@@ -37,9 +38,25 @@ object RuntimeConfig {
.createWithDefault("/tmp/libch")
// scalastyle:on line.size.limit
+ // scalastyle:off line.size.limit
+ val LOGGER_LEVEL =
+ buildConf(runtimeConfig("logger.level"))
+ .doc(
+ "https://clickhouse.com/docs/en/operations/server-configuration-parameters/settings#logger")
+ .stringConf
+ .createWithDefault("warning")
+ // scalastyle:on line.size.limit
+
+ /** Gluten Configuration */
val USE_CURRENT_DIRECTORY_AS_TMP =
buildConf(runtimeConfig("use_current_directory_as_tmp"))
.doc("Use the current directory as the temporary directory.")
.booleanConf
.createWithDefault(false)
+
+ val DUMP_PIPELINE =
+ buildConf(runtimeConfig("dump_pipeline"))
+ .doc("Dump pipeline to file after execution")
+ .booleanConf
+ .createWithDefault(false)
}
diff --git a/backends-clickhouse/src/main/scala/org/apache/spark/sql/execution/CHColumnarWrite.scala b/backends-clickhouse/src/main/scala/org/apache/spark/sql/execution/CHColumnarWrite.scala
index 1342e25043..427db0aad2 100644
--- a/backends-clickhouse/src/main/scala/org/apache/spark/sql/execution/CHColumnarWrite.scala
+++ b/backends-clickhouse/src/main/scala/org/apache/spark/sql/execution/CHColumnarWrite.scala
@@ -198,12 +198,12 @@ case class NativeStatCompute(rows: Seq[InternalRow]) {
}
case class NativeBasicWriteTaskStatsTracker(
- description: WriteJobDescription,
+ writeDir: String,
basicWriteJobStatsTracker: WriteTaskStatsTracker)
extends (NativeFileWriteResult => Unit) {
private var numWrittenRows: Long = 0
override def apply(stat: NativeFileWriteResult): Unit = {
- val absolutePath = s"${description.path}/${stat.relativePath}"
+ val absolutePath = s"$writeDir/${stat.relativePath}"
if (stat.partition_id != "__NO_PARTITION_ID__") {
basicWriteJobStatsTracker.newPartition(new GenericInternalRow(Array[Any](stat.partition_id)))
}
@@ -248,6 +248,8 @@ case class HadoopMapReduceCommitProtocolWrite(
extends CHColumnarWrite[HadoopMapReduceCommitProtocol]
with Logging {
+ private var stageDir: String = _
+
private lazy val adapter: HadoopMapReduceAdapter = HadoopMapReduceAdapter(committer)
/**
@@ -257,11 +259,12 @@ case class HadoopMapReduceCommitProtocolWrite(
override def doSetupNativeTask(): Unit = {
val (writePath, writeFilePattern) =
adapter.getTaskAttemptTempPathAndFilePattern(taskAttemptContext, description)
- logDebug(s"Native staging write path: $writePath and file pattern: $writeFilePattern")
+ stageDir = writePath
+ logDebug(s"Native staging write path: $stageDir and file pattern: $writeFilePattern")
val settings =
Map(
- RuntimeSettings.TASK_WRITE_TMP_DIR.key -> writePath,
+ RuntimeSettings.TASK_WRITE_TMP_DIR.key -> stageDir,
RuntimeSettings.TASK_WRITE_FILENAME_PATTERN.key -> writeFilePattern)
NativeExpressionEvaluator.updateQueryRuntimeSettings(settings)
}
@@ -272,7 +275,7 @@ case class HadoopMapReduceCommitProtocolWrite(
None
} else {
val commitInfo = FileCommitInfo(description)
- val basicNativeStat = NativeBasicWriteTaskStatsTracker(description, basicWriteJobStatsTracker)
+ val basicNativeStat = NativeBasicWriteTaskStatsTracker(stageDir, basicWriteJobStatsTracker)
val basicNativeStats = Seq(commitInfo, basicNativeStat)
NativeStatCompute(stats)(basicNativeStats)
val (partitions, addedAbsPathFiles) = commitInfo.result
diff --git a/backends-clickhouse/src/test/scala/org/apache/gluten/execution/GlutenClickHouseDeltaParquetWriteSuite.scala b/backends-clickhouse/src/test/scala/org/apache/gluten/execution/GlutenClickHouseDeltaParquetWriteSuite.scala
index 2f55510a7b..3736f0f144 100644
--- a/backends-clickhouse/src/test/scala/org/apache/gluten/execution/GlutenClickHouseDeltaParquetWriteSuite.scala
+++ b/backends-clickhouse/src/test/scala/org/apache/gluten/execution/GlutenClickHouseDeltaParquetWriteSuite.scala
@@ -1025,6 +1025,7 @@ class GlutenClickHouseDeltaParquetWriteSuite
}
}
+ // FIXME: optimize
testSparkVersionLE33("test parquet optimize with the path based table") {
val dataPath = s"$basePath/lineitem_delta_parquet_optimize_path_based"
clearDataPath(dataPath)
diff --git a/backends-clickhouse/src/test/scala/org/apache/gluten/execution/mergetree/GlutenClickHouseMergeTreeWriteSuite.scala b/backends-clickhouse/src/test/scala/org/apache/gluten/execution/mergetree/GlutenClickHouseMergeTreeWriteSuite.scala
index cc57760965..60ca58d9fc 100644
--- a/backends-clickhouse/src/test/scala/org/apache/gluten/execution/mergetree/GlutenClickHouseMergeTreeWriteSuite.scala
+++ b/backends-clickhouse/src/test/scala/org/apache/gluten/execution/mergetree/GlutenClickHouseMergeTreeWriteSuite.scala
@@ -57,6 +57,7 @@ class GlutenClickHouseMergeTreeWriteSuite
.set("spark.sql.adaptive.enabled", "true")
.set("spark.sql.files.maxPartitionBytes", "20000000")
.set(GlutenConfig.NATIVE_WRITER_ENABLED.key, "true")
+ .set(CHConf.ENABLE_ONEPIPELINE_MERGETREE_WRITE.key, spark35.toString)
.setCHSettings("min_insert_block_size_rows", 100000)
.setCHSettings("mergetree.merge_after_insert", false)
.setCHSettings("input_format_parquet_max_block_size", 8192)
@@ -67,178 +68,172 @@ class GlutenClickHouseMergeTreeWriteSuite
}
test("test mergetree table write") {
- withSQLConf((CHConf.ENABLE_ONEPIPELINE_MERGETREE_WRITE.key, spark35.toString)) {
- spark.sql(s"""
- |DROP TABLE IF EXISTS lineitem_mergetree;
- |""".stripMargin)
+ spark.sql(s"""
+ |DROP TABLE IF EXISTS lineitem_mergetree;
+ |""".stripMargin)
- // write.format.default = mergetree
- spark.sql(s"""
- |CREATE TABLE IF NOT EXISTS lineitem_mergetree
- |(
- | l_orderkey bigint,
- | l_partkey bigint,
- | l_suppkey bigint,
- | l_linenumber bigint,
- | l_quantity double,
- | l_extendedprice double,
- | l_discount double,
- | l_tax double,
- | l_returnflag string,
- | l_linestatus string,
- | l_shipdate date,
- | l_commitdate date,
- | l_receiptdate date,
- | l_shipinstruct string,
- | l_shipmode string,
- | l_comment string
- |)
- |USING clickhouse
- |TBLPROPERTIES (write.format.default = 'mergetree')
- |LOCATION '$basePath/lineitem_mergetree'
- |""".stripMargin)
+ // write.format.default = mergetree
+ spark.sql(s"""
+ |CREATE TABLE IF NOT EXISTS lineitem_mergetree
+ |(
+ | l_orderkey bigint,
+ | l_partkey bigint,
+ | l_suppkey bigint,
+ | l_linenumber bigint,
+ | l_quantity double,
+ | l_extendedprice double,
+ | l_discount double,
+ | l_tax double,
+ | l_returnflag string,
+ | l_linestatus string,
+ | l_shipdate date,
+ | l_commitdate date,
+ | l_receiptdate date,
+ | l_shipinstruct string,
+ | l_shipmode string,
+ | l_comment string
+ |)
+ |USING clickhouse
+ |TBLPROPERTIES (write.format.default = 'mergetree')
+ |LOCATION '$basePath/lineitem_mergetree'
+ |""".stripMargin)
- spark.sql(s"""
- | insert into table lineitem_mergetree
- | select * from lineitem
- |""".stripMargin)
+ spark.sql(s"""
+ | insert into table lineitem_mergetree
+ | select * from lineitem
+ |""".stripMargin)
- runTPCHQueryBySQL(1, q1("lineitem_mergetree")) {
- df =>
- val plans = collect(df.queryExecution.executedPlan) {
- case f: FileSourceScanExecTransformer => f
- case w: WholeStageTransformer => w
- }
- assertResult(4)(plans.size)
+ runTPCHQueryBySQL(1, q1("lineitem_mergetree")) {
+ df =>
+ val plans = collect(df.queryExecution.executedPlan) {
+ case f: FileSourceScanExecTransformer => f
+ case w: WholeStageTransformer => w
+ }
+ assertResult(4)(plans.size)
- val mergetreeScan = plans(3).asInstanceOf[FileSourceScanExecTransformer]
- assert(mergetreeScan.nodeName.startsWith("ScanTransformer mergetree"))
+ val mergetreeScan = plans(3).asInstanceOf[FileSourceScanExecTransformer]
+ assert(mergetreeScan.nodeName.startsWith("ScanTransformer mergetree"))
- val fileIndex = mergetreeScan.relation.location.asInstanceOf[TahoeFileIndex]
- assert(ClickHouseTableV2.getTable(fileIndex.deltaLog).clickhouseTableConfigs.nonEmpty)
- assert(ClickHouseTableV2.getTable(fileIndex.deltaLog).bucketOption.isEmpty)
- assert(ClickHouseTableV2.getTable(fileIndex.deltaLog).orderByKeyOption.isEmpty)
- assert(ClickHouseTableV2.getTable(fileIndex.deltaLog).primaryKeyOption.isEmpty)
- assert(ClickHouseTableV2.getTable(fileIndex.deltaLog).partitionColumns.isEmpty)
- val addFiles =
- fileIndex.matchingFiles(Nil, Nil).map(f => f.asInstanceOf[AddMergeTreeParts])
- assertResult(6)(addFiles.size)
- assertResult(600572)(addFiles.map(_.rows).sum)
+ val fileIndex = mergetreeScan.relation.location.asInstanceOf[TahoeFileIndex]
+ assert(ClickHouseTableV2.getTable(fileIndex.deltaLog).clickhouseTableConfigs.nonEmpty)
+ assert(ClickHouseTableV2.getTable(fileIndex.deltaLog).bucketOption.isEmpty)
+ assert(ClickHouseTableV2.getTable(fileIndex.deltaLog).orderByKeyOption.isEmpty)
+ assert(ClickHouseTableV2.getTable(fileIndex.deltaLog).primaryKeyOption.isEmpty)
+ assert(ClickHouseTableV2.getTable(fileIndex.deltaLog).partitionColumns.isEmpty)
+ val addFiles =
+ fileIndex.matchingFiles(Nil, Nil).map(f => f.asInstanceOf[AddMergeTreeParts])
+ assertResult(6)(addFiles.size)
+ assertResult(600572)(addFiles.map(_.rows).sum)
- // GLUTEN-5060: check the unnecessary FilterExec
- val wholeStageTransformer = plans(2).asInstanceOf[WholeStageTransformer]
- val planNodeJson = wholeStageTransformer.substraitPlanJson
- assert(
- !planNodeJson
- .replaceAll("\n", "")
- .replaceAll(" ", "")
- .contains("\"input\":{\"filter\":{"))
- }
+ // GLUTEN-5060: check the unnecessary FilterExec
+ val wholeStageTransformer = plans(2).asInstanceOf[WholeStageTransformer]
+ val planNodeJson = wholeStageTransformer.substraitPlanJson
+ assert(
+ !planNodeJson
+ .replaceAll("\n", "")
+ .replaceAll(" ", "")
+ .contains("\"input\":{\"filter\":{"))
}
}
test("test mergetree insert overwrite") {
- withSQLConf((CHConf.ENABLE_ONEPIPELINE_MERGETREE_WRITE.key, spark35.toString)) {
- spark.sql(s"""
- |DROP TABLE IF EXISTS lineitem_mergetree_insertoverwrite;
- |""".stripMargin)
+ spark.sql(s"""
+ |DROP TABLE IF EXISTS lineitem_mergetree_insertoverwrite;
+ |""".stripMargin)
- spark.sql(s"""
- |CREATE TABLE IF NOT EXISTS lineitem_mergetree_insertoverwrite
- |(
- | l_orderkey bigint,
- | l_partkey bigint,
- | l_suppkey bigint,
- | l_linenumber bigint,
- | l_quantity double,
- | l_extendedprice double,
- | l_discount double,
- | l_tax double,
- | l_returnflag string,
- | l_linestatus string,
- | l_shipdate date,
- | l_commitdate date,
- | l_receiptdate date,
- | l_shipinstruct string,
- | l_shipmode string,
- | l_comment string
- |)
- |USING clickhouse
- |LOCATION '$basePath/lineitem_mergetree_insertoverwrite'
- |""".stripMargin)
+ spark.sql(s"""
+ |CREATE TABLE IF NOT EXISTS lineitem_mergetree_insertoverwrite
+ |(
+ | l_orderkey bigint,
+ | l_partkey bigint,
+ | l_suppkey bigint,
+ | l_linenumber bigint,
+ | l_quantity double,
+ | l_extendedprice double,
+ | l_discount double,
+ | l_tax double,
+ | l_returnflag string,
+ | l_linestatus string,
+ | l_shipdate date,
+ | l_commitdate date,
+ | l_receiptdate date,
+ | l_shipinstruct string,
+ | l_shipmode string,
+ | l_comment string
+ |)
+ |USING clickhouse
+ |LOCATION '$basePath/lineitem_mergetree_insertoverwrite'
+ |""".stripMargin)
- spark.sql(s"""
- | insert into table lineitem_mergetree_insertoverwrite
- | select * from lineitem
- |""".stripMargin)
+ spark.sql(s"""
+ | insert into table lineitem_mergetree_insertoverwrite
+ | select * from lineitem
+ |""".stripMargin)
- spark.sql(s"""
- | insert overwrite table lineitem_mergetree_insertoverwrite
- | select * from lineitem where mod(l_orderkey,2) = 1
- |""".stripMargin)
- val sql2 =
- s"""
- | select count(*) from lineitem_mergetree_insertoverwrite
- |
- |""".stripMargin
- assertResult(300001)(
- // total rows should remain unchanged
- spark.sql(sql2).collect().apply(0).get(0)
- )
- }
+ spark.sql(s"""
+ | insert overwrite table lineitem_mergetree_insertoverwrite
+ | select * from lineitem where mod(l_orderkey,2) = 1
+ |""".stripMargin)
+ val sql2 =
+ s"""
+ | select count(*) from lineitem_mergetree_insertoverwrite
+ |
+ |""".stripMargin
+ assertResult(300001)(
+ // total rows should remain unchanged
+ spark.sql(sql2).collect().apply(0).get(0)
+ )
}
test("test mergetree insert overwrite partitioned table with small table, static") {
- withSQLConf((CHConf.ENABLE_ONEPIPELINE_MERGETREE_WRITE.key, spark35.toString)) {
- spark.sql(s"""
- |DROP TABLE IF EXISTS lineitem_mergetree_insertoverwrite2;
- |""".stripMargin)
+ spark.sql(s"""
+ |DROP TABLE IF EXISTS lineitem_mergetree_insertoverwrite2;
+ |""".stripMargin)
- spark.sql(s"""
- |CREATE TABLE IF NOT EXISTS lineitem_mergetree_insertoverwrite2
- |(
- | l_orderkey bigint,
- | l_partkey bigint,
- | l_suppkey bigint,
- | l_linenumber bigint,
- | l_quantity double,
- | l_extendedprice double,
- | l_discount double,
- | l_tax double,
- | l_returnflag string,
- | l_linestatus string,
- | l_shipdate date,
- | l_commitdate date,
- | l_receiptdate date,
- | l_shipinstruct string,
- | l_shipmode string,
- | l_comment string
- |)
- |USING clickhouse
- |PARTITIONED BY (l_shipdate)
- |LOCATION '$basePath/lineitem_mergetree_insertoverwrite2'
- |""".stripMargin)
+ spark.sql(s"""
+ |CREATE TABLE IF NOT EXISTS lineitem_mergetree_insertoverwrite2
+ |(
+ | l_orderkey bigint,
+ | l_partkey bigint,
+ | l_suppkey bigint,
+ | l_linenumber bigint,
+ | l_quantity double,
+ | l_extendedprice double,
+ | l_discount double,
+ | l_tax double,
+ | l_returnflag string,
+ | l_linestatus string,
+ | l_shipdate date,
+ | l_commitdate date,
+ | l_receiptdate date,
+ | l_shipinstruct string,
+ | l_shipmode string,
+ | l_comment string
+ |)
+ |USING clickhouse
+ |PARTITIONED BY (l_shipdate)
+ |LOCATION '$basePath/lineitem_mergetree_insertoverwrite2'
+ |""".stripMargin)
- spark.sql(s"""
- | insert into table lineitem_mergetree_insertoverwrite2
- | select * from lineitem
- |""".stripMargin)
+ spark.sql(s"""
+ | insert into table lineitem_mergetree_insertoverwrite2
+ | select * from lineitem
+ |""".stripMargin)
- spark.sql(
- s"""
- | insert overwrite table lineitem_mergetree_insertoverwrite2
- | select * from lineitem where l_shipdate BETWEEN date'1993-02-01' AND date'1993-02-10'
- |""".stripMargin)
- val sql2 =
- s"""
- | select count(*) from lineitem_mergetree_insertoverwrite2
- |
- |""".stripMargin
- assertResult(2418)(
- // total rows should remain unchanged
- spark.sql(sql2).collect().apply(0).get(0)
- )
- }
+ spark.sql(
+ s"""
+ | insert overwrite table lineitem_mergetree_insertoverwrite2
+ | select * from lineitem where l_shipdate BETWEEN date'1993-02-01' AND date'1993-02-10'
+ |""".stripMargin)
+ val sql2 =
+ s"""
+ | select count(*) from lineitem_mergetree_insertoverwrite2
+ |
+ |""".stripMargin
+ assertResult(2418)(
+ // total rows should remain unchanged
+ spark.sql(sql2).collect().apply(0).get(0)
+ )
}
test("test mergetree insert overwrite partitioned table with small table, dynamic") {
@@ -650,8 +645,8 @@ class GlutenClickHouseMergeTreeWriteSuite
// static partition
spark.sql(s"""
- | insert into lineitem_mergetree_partition PARTITION (l_shipdate=date'1995-01-21',
- | l_returnflag = 'A')
+ | insert into lineitem_mergetree_partition
+ | PARTITION (l_shipdate=date'1995-01-21', l_returnflag = 'A')
| (l_orderkey,
| l_partkey,
| l_suppkey,
@@ -729,7 +724,8 @@ class GlutenClickHouseMergeTreeWriteSuite
ClickHouseTableV2
.getTable(fileIndex.deltaLog)
.partitionColumns(1))
- val addFiles = fileIndex.matchingFiles(Nil, Nil).map(f => f.asInstanceOf[AddMergeTreeParts])
+ val addFiles =
+ fileIndex.matchingFiles(Nil, Nil).map(f => f.asInstanceOf[AddMergeTreeParts])
assertResult(3836)(addFiles.size)
assertResult(605363)(addFiles.map(_.rows).sum)
@@ -739,7 +735,7 @@ class GlutenClickHouseMergeTreeWriteSuite
}
}
- test("test mergetree write with bucket table") {
+ testSparkVersionLE33("test mergetree write with bucket table") {
spark.sql(s"""
|DROP TABLE IF EXISTS lineitem_mergetree_bucket;
|""".stripMargin)
@@ -979,7 +975,7 @@ class GlutenClickHouseMergeTreeWriteSuite
}
}
- test("test mergetree CTAS complex") {
+ test("test mergetree CTAS partition") {
spark.sql(s"""
|DROP TABLE IF EXISTS lineitem_mergetree_ctas2;
|""".stripMargin)
@@ -988,8 +984,6 @@ class GlutenClickHouseMergeTreeWriteSuite
|CREATE TABLE IF NOT EXISTS lineitem_mergetree_ctas2
|USING clickhouse
|PARTITIONED BY (l_shipdate)
- |CLUSTERED BY (l_orderkey)
- |${if (spark32) "" else "SORTED BY (l_partkey, l_returnflag)"} INTO 4 BUCKETS
|LOCATION '$basePath/lineitem_mergetree_ctas2'
| as select * from lineitem
|""".stripMargin)
@@ -1598,7 +1592,7 @@ class GlutenClickHouseMergeTreeWriteSuite
case scanExec: BasicScanExecTransformer => scanExec
}
assertResult(1)(plans.size)
- assertResult(conf._2)(plans.head.getSplitInfos.size)
+ assertResult(conf._2)(plans.head.getSplitInfos().size)
}
}
})
@@ -1622,12 +1616,12 @@ class GlutenClickHouseMergeTreeWriteSuite
case scanExec: BasicScanExecTransformer => scanExec
}
assertResult(1)(plans.size)
- assertResult(1)(plans.head.getSplitInfos.size)
+ assertResult(1)(plans.head.getSplitInfos().size)
}
}
}
- test("test mergetree with primary keys filter pruning by driver with bucket") {
+ testSparkVersionLE33("test mergetree with primary keys filter pruning by driver with bucket") {
spark.sql(s"""
|DROP TABLE IF EXISTS lineitem_mergetree_pk_pruning_by_driver_bucket;
|""".stripMargin)
@@ -1730,7 +1724,7 @@ class GlutenClickHouseMergeTreeWriteSuite
case f: BasicScanExecTransformer => f
}
assertResult(2)(scanExec.size)
- assertResult(conf._2)(scanExec(1).getSplitInfos.size)
+ assertResult(conf._2)(scanExec(1).getSplitInfos().size)
}
}
})
@@ -1776,7 +1770,7 @@ class GlutenClickHouseMergeTreeWriteSuite
Seq("true", "false").foreach {
skip =>
- withSQLConf("spark.databricks.delta.stats.skipping" -> skip.toString) {
+ withSQLConf("spark.databricks.delta.stats.skipping" -> skip) {
val sqlStr =
s"""
|SELECT
@@ -1799,7 +1793,7 @@ class GlutenClickHouseMergeTreeWriteSuite
}
}
- test("test mergetree with column case sensitive") {
+ testSparkVersionLE33("test mergetree with column case sensitive") {
spark.sql(s"""
|DROP TABLE IF EXISTS LINEITEM_MERGETREE_CASE_SENSITIVE;
|""".stripMargin)
@@ -1838,7 +1832,7 @@ class GlutenClickHouseMergeTreeWriteSuite
runTPCHQueryBySQL(6, q6("lineitem_mergetree_case_sensitive")) { _ => }
}
- test("test mergetree with partition with whitespace") {
+ testSparkVersionLE33("test mergetree with partition with whitespace") {
spark.sql(s"""
|DROP TABLE IF EXISTS lineitem_mergetree_partition_with_whitespace;
|""".stripMargin)
@@ -1900,7 +1894,7 @@ class GlutenClickHouseMergeTreeWriteSuite
Seq(("-1", 3), ("3", 3), ("6", 1)).foreach(
conf => {
withSQLConf(
- ("spark.gluten.sql.columnar.backend.ch.files.per.partition.threshold" -> conf._1)) {
+ "spark.gluten.sql.columnar.backend.ch.files.per.partition.threshold" -> conf._1) {
val sql =
s"""
|select count(1), min(l_returnflag) from lineitem_split
@@ -1913,7 +1907,7 @@ class GlutenClickHouseMergeTreeWriteSuite
val scanExec = collect(df.queryExecution.executedPlan) {
case f: FileSourceScanExecTransformer => f
}
- assert(scanExec(0).getPartitions.size == conf._2)
+ assert(scanExec.head.getPartitions.size == conf._2)
}
}
})
diff --git a/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxTransformerApi.scala b/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxTransformerApi.scala
index c6d2bc0658..d156fffa8b 100644
--- a/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxTransformerApi.scala
+++ b/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxTransformerApi.scala
@@ -27,7 +27,7 @@ import org.apache.gluten.vectorized.PlanEvaluatorJniWrapper
import org.apache.spark.internal.Logging
import org.apache.spark.sql.catalyst.expressions.{Attribute, Expression}
import org.apache.spark.sql.connector.read.InputPartition
-import org.apache.spark.sql.execution.datasources.{FileFormat, HadoopFsRelation, PartitionDirectory}
+import org.apache.spark.sql.execution.datasources.{HadoopFsRelation, PartitionDirectory}
import org.apache.spark.sql.sources.DataSourceRegister
import org.apache.spark.sql.types._
import org.apache.spark.task.TaskResources
@@ -96,16 +96,14 @@ class VeloxTransformerApi extends TransformerApi with Logging {
override def packPBMessage(message: Message): Any = Any.pack(message, "")
- override def genWriteParameters(
- fileFormat: FileFormat,
- writeOptions: Map[String, String]): Any = {
- val fileFormatStr = fileFormat match {
+ override def genWriteParameters(write: WriteFilesExecTransformer): Any = {
+ val fileFormatStr = write.fileFormat match {
case register: DataSourceRegister =>
register.shortName
case _ => "UnknownFileFormat"
}
val compressionCodec =
- WriteFilesExecTransformer.getCompressionCodec(writeOptions).capitalize
+ WriteFilesExecTransformer.getCompressionCodec(write.caseInsensitiveOptions).capitalize
val writeParametersStr = new StringBuffer("WriteParameters:")
writeParametersStr.append("is").append(compressionCodec).append("=1")
writeParametersStr.append(";format=").append(fileFormatStr).append("\n")
diff --git a/cpp-ch/local-engine/Parser/RelParsers/WriteRelParser.cpp b/cpp-ch/local-engine/Parser/RelParsers/WriteRelParser.cpp
index a76b4d398d..0d57d53ff6 100644
--- a/cpp-ch/local-engine/Parser/RelParsers/WriteRelParser.cpp
+++ b/cpp-ch/local-engine/Parser/RelParsers/WriteRelParser.cpp
@@ -21,9 +21,8 @@
#include <DataTypes/DataTypeTuple.h>
#include <Interpreters/Context.h>
#include <Parser/TypeParser.h>
-#include <Processors/Transforms/ApplySquashingTransform.h>
#include <Processors/Transforms/ExpressionTransform.h>
-#include <Processors/Transforms/PlanSquashingTransform.h>
+#include <Processors/Transforms/MaterializingTransform.h>
#include <QueryPipeline/QueryPipelineBuilder.h>
#include <Storages/MergeTree/SparkMergeTreeMeta.h>
#include <Storages/MergeTree/SparkMergeTreeSink.h>
@@ -103,7 +102,7 @@ void adjust_output(const DB::QueryPipelineBuilderPtr & builder, const DB::Block
{
throw DB::Exception(
DB::ErrorCodes::LOGICAL_ERROR,
- "Missmatch result columns size, input size is {}, but output size is {}",
+ "Mismatch result columns size, input size is {}, but output size is {}",
input.columns(),
output.columns());
}
@@ -164,12 +163,6 @@ void addMergeTreeSinkTransform(
: std::make_shared<SparkMergeTreePartitionedFileSink>(header, partition_by, merge_tree_table, write_settings, context, stats);
chain.addSource(sink);
- const DB::Settings & settings = context->getSettingsRef();
- chain.addSource(std::make_shared<ApplySquashingTransform>(
- header, settings[Setting::min_insert_block_size_rows], settings[Setting::min_insert_block_size_bytes]));
- chain.addSource(std::make_shared<PlanSquashingTransform>(
- header, settings[Setting::min_insert_block_size_rows], settings[Setting::min_insert_block_size_bytes]));
-
builder->addChain(std::move(chain));
}
@@ -212,6 +205,7 @@ void addNormalFileWriterSinkTransform(
namespace local_engine
{
+
IMPLEMENT_GLUTEN_SETTINGS(GlutenWriteSettings, WRITE_RELATED_SETTINGS)
void addSinkTransform(const DB::ContextPtr & context, const substrait::WriteRel & write_rel, const DB::QueryPipelineBuilderPtr & builder)
@@ -224,12 +218,18 @@ void addSinkTransform(const DB::ContextPtr & context, const substrait::WriteRel
throw DB::Exception(DB::ErrorCodes::LOGICAL_ERROR, "Failed to unpack write optimization with local_engine::Write.");
assert(write.has_common());
const substrait::NamedStruct & table_schema = write_rel.table_schema();
- auto output = TypeParser::buildBlockFromNamedStruct(table_schema);
- adjust_output(builder, output);
- const auto partitionCols = collect_partition_cols(output, table_schema);
+ auto partition_indexes = write.common().partition_col_index();
if (write.has_mergetree())
{
- local_engine::MergeTreeTable merge_tree_table(write, table_schema);
+ MergeTreeTable merge_tree_table(write, table_schema);
+ auto output = TypeParser::buildBlockFromNamedStruct(table_schema, merge_tree_table.low_card_key);
+ adjust_output(builder, output);
+
+ builder->addSimpleTransform(
+ [&](const Block & in_header) -> ProcessorPtr { return std::make_shared<MaterializingTransform>(in_header, false); });
+
+ const auto partition_by = collect_partition_cols(output, table_schema, partition_indexes);
+
GlutenWriteSettings write_settings = GlutenWriteSettings::get(context);
if (write_settings.task_write_tmp_dir.empty())
throw DB::Exception(DB::ErrorCodes::LOGICAL_ERROR, "MergeTree Write Pipeline need inject relative path.");
@@ -237,23 +237,35 @@ void addSinkTransform(const DB::ContextPtr & context, const substrait::WriteRel
throw DB::Exception(DB::ErrorCodes::LOGICAL_ERROR, "Non empty relative path for MergeTree table in pipeline mode.");
merge_tree_table.relative_path = write_settings.task_write_tmp_dir;
- addMergeTreeSinkTransform(context, builder, merge_tree_table, output, partitionCols);
+ addMergeTreeSinkTransform(context, builder, merge_tree_table, output, partition_by);
}
else
- addNormalFileWriterSinkTransform(context, builder, write.common().format(), output, partitionCols);
+ {
+ auto output = TypeParser::buildBlockFromNamedStruct(table_schema);
+ adjust_output(builder, output);
+ const auto partition_by = collect_partition_cols(output, table_schema, partition_indexes);
+ addNormalFileWriterSinkTransform(context, builder, write.common().format(), output, partition_by);
+ }
}
-
-DB::Names collect_partition_cols(const DB::Block & header, const substrait::NamedStruct & struct_)
+DB::Names collect_partition_cols(const DB::Block & header, const substrait::NamedStruct & struct_, const PartitionIndexes & partition_by)
{
- DB::Names result;
+ if (partition_by.empty())
+ {
+ assert(std::ranges::all_of(
+ struct_.column_types(), [](const int32_t type) { return type != ::substrait::NamedStruct::PARTITION_COL; }));
+ return {};
+ }
assert(struct_.column_types_size() == header.columns());
assert(struct_.column_types_size() == struct_.struct_().types_size());
- auto name_iter = header.begin();
- auto type_iter = struct_.column_types().begin();
- for (; name_iter != header.end(); ++name_iter, ++type_iter)
- if (*type_iter == ::substrait::NamedStruct::PARTITION_COL)
- result.push_back(name_iter->name);
+ DB::Names result;
+ result.reserve(partition_by.size());
+ for (auto idx : partition_by)
+ {
+ assert(idx >= 0 && idx < header.columns());
+ assert(struct_.column_types(idx) == ::substrait::NamedStruct::PARTITION_COL);
+ result.emplace_back(header.getByPosition(idx).name);
+ }
return result;
}
diff --git a/cpp-ch/local-engine/Parser/RelParsers/WriteRelParser.h b/cpp-ch/local-engine/Parser/RelParsers/WriteRelParser.h
index 01e0dabaaa..bb8c15c07d 100644
--- a/cpp-ch/local-engine/Parser/RelParsers/WriteRelParser.h
+++ b/cpp-ch/local-engine/Parser/RelParsers/WriteRelParser.h
@@ -21,6 +21,7 @@
#include <Core/Block.h>
#include <Core/Names.h>
#include <Interpreters/Context_fwd.h>
+#include <google/protobuf/repeated_field.h>
#include <Common/GlutenSettings.h>
namespace substrait
@@ -38,9 +39,11 @@ using QueryPipelineBuilderPtr = std::unique_ptr<QueryPipelineBuilder>;
namespace local_engine
{
+using PartitionIndexes = google::protobuf::RepeatedField<::int32_t>;
+
void addSinkTransform(const DB::ContextPtr & context, const substrait::WriteRel & write_rel, const DB::QueryPipelineBuilderPtr & builder);
-DB::Names collect_partition_cols(const DB::Block & header, const substrait::NamedStruct & struct_);
+DB::Names collect_partition_cols(const DB::Block & header, const substrait::NamedStruct & struct_, const PartitionIndexes & partition_by);
#define WRITE_RELATED_SETTINGS(M, ALIAS) \
M(String, task_write_tmp_dir, , "The temporary directory for writing data") \
diff --git a/cpp-ch/local-engine/Storages/MergeTree/SparkMergeTreeSink.cpp b/cpp-ch/local-engine/Storages/MergeTree/SparkMergeTreeSink.cpp
index 6c9dd890d8..d41e71fb84 100644
--- a/cpp-ch/local-engine/Storages/MergeTree/SparkMergeTreeSink.cpp
+++ b/cpp-ch/local-engine/Storages/MergeTree/SparkMergeTreeSink.cpp
@@ -31,27 +31,37 @@ extern const Metric GlobalThreadActive;
extern const Metric GlobalThreadScheduled;
}
+namespace DB::Setting
+{
+extern const SettingsUInt64 min_insert_block_size_rows;
+extern const SettingsUInt64 min_insert_block_size_bytes;
+}
namespace local_engine
{
-void SparkMergeTreeSink::consume(Chunk & chunk)
+void SparkMergeTreeSink::write(const Chunk & chunk)
{
- assert(!sink_helper->metadata_snapshot->hasPartitionKey());
+ CurrentThread::flushUntrackedMemory();
+ /// Reset earlier, so put it in the scope
BlockWithPartition item{getHeader().cloneWithColumns(chunk.getColumns()), Row{}};
- size_t before_write_memory = 0;
- if (auto * memory_tracker = CurrentThread::getMemoryTracker())
- {
- CurrentThread::flushUntrackedMemory();
- before_write_memory = memory_tracker->get();
- }
+
sink_helper->writeTempPart(item, context, part_num);
part_num++;
- /// Reset earlier to free memory
- item.block.clear();
- item.partition.clear();
+}
- sink_helper->checkAndMerge();
+void SparkMergeTreeSink::consume(Chunk & chunk)
+{
+ Chunk tmp;
+ tmp.swap(chunk);
+ squashed_chunk = squashing.add(std::move(tmp));
+ if (static_cast<bool>(squashed_chunk))
+ {
+ write(Squashing::squash(std::move(squashed_chunk)));
+ sink_helper->checkAndMerge();
+ }
+ assert(squashed_chunk.getNumRows() == 0);
+ assert(chunk.getNumRows() == 0);
}
void SparkMergeTreeSink::onStart()
@@ -61,6 +71,11 @@ void SparkMergeTreeSink::onStart()
void SparkMergeTreeSink::onFinish()
{
+ assert(squashed_chunk.getNumRows() == 0);
+ squashed_chunk = squashing.flush();
+ if (static_cast<bool>(squashed_chunk))
+ write(Squashing::squash(std::move(squashed_chunk)));
+ assert(squashed_chunk.getNumRows() == 0);
sink_helper->finish(context);
if (stats_.has_value())
(*stats_)->collectStats(sink_helper->unsafeGet(), sink_helper->write_settings.partition_settings.partition_dir);
@@ -91,7 +106,9 @@ SinkToStoragePtr SparkMergeTreeSink::create(
}
else
sink_helper = std::make_shared<DirectSinkHelper>(dest_storage, write_settings_, isRemoteStorage);
- return std::make_shared<SparkMergeTreeSink>(sink_helper, context, stats);
+ const DB::Settings & settings = context->getSettingsRef();
+ return std::make_shared<SparkMergeTreeSink>(
+ sink_helper, context, stats, settings[Setting::min_insert_block_size_rows], settings[Setting::min_insert_block_size_bytes]);
}
SinkHelper::SinkHelper(const SparkStorageMergeTreePtr & data_, const SparkMergeTreeWriteSettings & write_settings_, bool isRemoteStorage_)
diff --git a/cpp-ch/local-engine/Storages/MergeTree/SparkMergeTreeSink.h b/cpp-ch/local-engine/Storages/MergeTree/SparkMergeTreeSink.h
index b551d86d1d..828332d2d6 100644
--- a/cpp-ch/local-engine/Storages/MergeTree/SparkMergeTreeSink.h
+++ b/cpp-ch/local-engine/Storages/MergeTree/SparkMergeTreeSink.h
@@ -227,8 +227,17 @@ public:
const DB::ContextMutablePtr & context,
const SinkStatsOption & stats = {});
- explicit SparkMergeTreeSink(const SinkHelperPtr & sink_helper_, const ContextPtr & context_, const SinkStatsOption & stats)
- : SinkToStorage(sink_helper_->metadata_snapshot->getSampleBlock()), context(context_), sink_helper(sink_helper_), stats_(stats)
+ explicit SparkMergeTreeSink(
+ const SinkHelperPtr & sink_helper_,
+ const ContextPtr & context_,
+ const SinkStatsOption & stats,
+ size_t min_block_size_rows,
+ size_t min_block_size_bytes)
+ : SinkToStorage(sink_helper_->metadata_snapshot->getSampleBlock())
+ , context(context_)
+ , sink_helper(sink_helper_)
+ , stats_(stats)
+ , squashing(sink_helper_->metadata_snapshot->getSampleBlock(), min_block_size_rows, min_block_size_bytes)
{
}
~SparkMergeTreeSink() override = default;
@@ -241,9 +250,13 @@ public:
const SinkHelper & sinkHelper() const { return *sink_helper; }
private:
+ void write(const Chunk & chunk);
+
ContextPtr context;
SinkHelperPtr sink_helper;
std::optional<std::shared_ptr<MergeTreeStats>> stats_;
+ Squashing squashing;
+ Chunk squashed_chunk;
int part_num = 1;
};
diff --git a/cpp-ch/local-engine/Storages/MergeTree/SparkMergeTreeWriter.cpp b/cpp-ch/local-engine/Storages/MergeTree/SparkMergeTreeWriter.cpp
index a8fdfff6ff..95145d43fa 100644
--- a/cpp-ch/local-engine/Storages/MergeTree/SparkMergeTreeWriter.cpp
+++ b/cpp-ch/local-engine/Storages/MergeTree/SparkMergeTreeWriter.cpp
@@ -18,8 +18,6 @@
#include <Core/Settings.h>
#include <Interpreters/ActionsDAG.h>
-#include <Processors/Transforms/ApplySquashingTransform.h>
-#include <Processors/Transforms/PlanSquashingTransform.h>
#include <Storages/MergeTree/DataPartStorageOnDiskFull.h>
#include <Storages/MergeTree/MetaDataHelper.h>
#include <Storages/MergeTree/SparkMergeTreeSink.h>
@@ -28,11 +26,6 @@
#include <Poco/StringTokenizer.h>
#include <Common/JNIUtils.h>
-namespace DB::Setting
-{
-extern const SettingsUInt64 min_insert_block_size_rows;
-extern const SettingsUInt64 min_insert_block_size_bytes;
-}
using namespace DB;
namespace
{
@@ -125,12 +118,6 @@ std::unique_ptr<SparkMergeTreeWriter> SparkMergeTreeWriter::create(
//
// auto stats = std::make_shared<MergeTreeStats>(header, sink_helper);
// chain.addSink(stats);
- //
- chain.addSource(std::make_shared<ApplySquashingTransform>(
- header, settings[Setting::min_insert_block_size_rows], settings[Setting::min_insert_block_size_bytes]));
- chain.addSource(std::make_shared<PlanSquashingTransform>(
- header, settings[Setting::min_insert_block_size_rows], settings[Setting::min_insert_block_size_bytes]));
-
return std::make_unique<SparkMergeTreeWriter>(header, sink_helper, QueryPipeline{std::move(chain)}, spark_job_id);
}
diff --git a/cpp-ch/local-engine/tests/gtest_write_pipeline.cpp b/cpp-ch/local-engine/tests/gtest_write_pipeline.cpp
index a01dd363c5..a36601d6af 100644
--- a/cpp-ch/local-engine/tests/gtest_write_pipeline.cpp
+++ b/cpp-ch/local-engine/tests/gtest_write_pipeline.cpp
@@ -146,7 +146,7 @@ TEST(WritePipeline, SubstraitFileSink)
DB::Names expected{"s_suppkey", "s_name", "s_address", "s_nationkey", "s_phone", "s_acctbal", "s_comment111"};
EXPECT_EQ(expected, names);
- auto partitionCols = collect_partition_cols(block, table_schema);
+ auto partitionCols = collect_partition_cols(block, table_schema, {});
DB::Names expected_partition_cols;
EXPECT_EQ(expected_partition_cols, partitionCols);
@@ -164,7 +164,7 @@ TEST(WritePipeline, SubstraitFileSink)
INCBIN(native_write_one_partition, SOURCE_DIR "/utils/extern-local-engine/tests/json/native_write_one_partition.json");
-TEST(WritePipeline, SubstraitPartitionedFileSink)
+/*TEST(WritePipeline, SubstraitPartitionedFileSink)
{
const auto context = DB::Context::createCopy(QueryContext::globalContext());
GlutenWriteSettings settings{
@@ -193,7 +193,7 @@ TEST(WritePipeline, SubstraitPartitionedFileSink)
DB::Names expected{"s_suppkey", "s_name", "s_address", "s_phone", "s_acctbal", "s_comment", "s_nationkey"};
EXPECT_EQ(expected, names);
- auto partitionCols = local_engine::collect_partition_cols(block, table_schema);
+ auto partitionCols = local_engine::collect_partition_cols(block, table_schema, {});
DB::Names expected_partition_cols{"s_nationkey"};
EXPECT_EQ(expected_partition_cols, partitionCols);
@@ -201,12 +201,12 @@ TEST(WritePipeline, SubstraitPartitionedFileSink)
const Block & x = *local_executor->nextColumnar();
debug::headBlock(x, 25);
EXPECT_EQ(25, x.rows());
-}
+}*/
TEST(WritePipeline, ComputePartitionedExpression)
{
const auto context = DB::Context::createCopy(QueryContext::globalContext());
-
+
Block sample_block{{STRING(), "name"}, {UINT(), "s_nationkey"}};
auto partition_by = SubstraitPartitionedFileSink::make_partition_expression({"s_nationkey", "name"}, sample_block);
// auto partition_by = printColumn("s_nationkey");
diff --git a/cpp-ch/local-engine/tests/gtest_write_pipeline_mergetree.cpp b/cpp-ch/local-engine/tests/gtest_write_pipeline_mergetree.cpp
index 1ad90060f4..a5cd3fd7f3 100644
--- a/cpp-ch/local-engine/tests/gtest_write_pipeline_mergetree.cpp
+++ b/cpp-ch/local-engine/tests/gtest_write_pipeline_mergetree.cpp
@@ -258,11 +258,18 @@ TEST(MergeTree, SparkMergeTree)
INCBIN(_3_mergetree_plan_input_, SOURCE_DIR "/utils/extern-local-engine/tests/json/mergetree/lineitem_parquet_input.json");
namespace
{
-void writeMerge(std::string_view json_plan,
- const std::string & outputPath ,
- const std::function<void(const DB::Block &)> & callback, std::optional<std::string> input = std::nullopt)
+void writeMerge(
+ std::string_view json_plan,
+ const std::string & outputPath,
+ const std::function<void(const DB::Block &)> & callback,
+ std::optional<std::string> input = std::nullopt)
{
const auto context = DB::Context::createCopy(QueryContext::globalContext());
+
+ auto queryid = QueryContext::instance().initializeQuery("gtest_mergetree");
+ SCOPE_EXIT({ QueryContext::instance().finalizeQuery(queryid); });
+
+
GlutenWriteSettings settings{.task_write_tmp_dir = outputPath};
settings.set(context);
SparkMergeTreeWritePartitionSettings partition_settings{.part_name_prefix = "pipline_prefix"};
@@ -279,18 +286,24 @@ INCBIN(_3_mergetree_plan_, SOURCE_DIR "/utils/extern-local-engine/tests/json/mer
INCBIN(_4_mergetree_plan_, SOURCE_DIR "/utils/extern-local-engine/tests/json/mergetree/4_one_pipeline.json");
TEST(MergeTree, Pipeline)
{
- writeMerge(EMBEDDED_PLAN(_3_mergetree_plan_),"tmp/lineitem_mergetree",[&](const DB::Block & block)
- {
- EXPECT_EQ(1, block.rows());
- debug::headBlock(block);
- });
+ writeMerge(
+ EMBEDDED_PLAN(_3_mergetree_plan_),
+ "tmp/lineitem_mergetree",
+ [&](const DB::Block & block)
+ {
+ EXPECT_EQ(1, block.rows());
+ debug::headBlock(block);
+ });
}
TEST(MergeTree, PipelineWithPartition)
{
- writeMerge(EMBEDDED_PLAN(_4_mergetree_plan_),"tmp/lineitem_mergetree_p",[&](const DB::Block & block)
- {
- EXPECT_EQ(2525, block.rows());
- debug::headBlock(block);
- });
+ writeMerge(
+ EMBEDDED_PLAN(_4_mergetree_plan_),
+ "tmp/lineitem_mergetree_p",
+ [&](const DB::Block & block)
+ {
+ EXPECT_EQ(3815, block.rows());
+ debug::headBlock(block);
+ });
}
\ No newline at end of file
diff --git a/cpp-ch/local-engine/tests/json/mergetree/4_one_pipeline.json b/cpp-ch/local-engine/tests/json/mergetree/4_one_pipeline.json
index 14a9b3dda2..513f54a707 100644
--- a/cpp-ch/local-engine/tests/json/mergetree/4_one_pipeline.json
+++ b/cpp-ch/local-engine/tests/json/mergetree/4_one_pipeline.json
@@ -9,13 +9,18 @@
"optimization": {
"@type": "type.googleapis.com/local_engine.Write",
"common": {
- "format": "mergetree"
+ "format": "mergetree",
+ "partitionColIndex": [
+ 10,
+ 8
+ ]
},
"mergetree": {
"database": "default",
- "table": "lineitem_mergetree_insertoverwrite2",
- "snapshotId": "1731309448915_0",
- "orderByKey": "tuple()",
+ "table": "lineitem_mergetree_partition",
+ "snapshotId": "1734145864855_0",
+ "orderByKey": "l_orderkey",
+ "primaryKey": "l_orderkey",
"storagePolicy": "default"
}
},
@@ -221,7 +226,7 @@
"NORMAL_COL",
"NORMAL_COL",
"NORMAL_COL",
- "NORMAL_COL",
+ "PARTITION_COL",
"NORMAL_COL",
"PARTITION_COL",
"NORMAL_COL",
@@ -232,138 +237,171 @@
]
},
"input": {
- "read": {
+ "sort": {
"common": {
"direct": {}
},
- "baseSchema": {
- "names": [
- "l_orderkey",
- "l_partkey",
- "l_suppkey",
- "l_linenumber",
- "l_quantity",
- "l_extendedprice",
- "l_discount",
- "l_tax",
- "l_returnflag",
- "l_linestatus",
- "l_shipdate",
- "l_commitdate",
- "l_receiptdate",
- "l_shipinstruct",
- "l_shipmode",
- "l_comment"
- ],
- "struct": {
- "types": [
- {
- "i64": {
- "nullability": "NULLABILITY_NULLABLE"
- }
- },
- {
- "i64": {
- "nullability": "NULLABILITY_NULLABLE"
- }
- },
- {
- "i64": {
- "nullability": "NULLABILITY_NULLABLE"
- }
- },
- {
- "i64": {
- "nullability": "NULLABILITY_NULLABLE"
- }
- },
- {
- "fp64": {
- "nullability": "NULLABILITY_NULLABLE"
- }
- },
- {
- "fp64": {
- "nullability": "NULLABILITY_NULLABLE"
- }
- },
- {
- "fp64": {
- "nullability": "NULLABILITY_NULLABLE"
- }
- },
- {
- "fp64": {
- "nullability": "NULLABILITY_NULLABLE"
- }
- },
- {
- "string": {
- "nullability": "NULLABILITY_NULLABLE"
- }
- },
- {
- "string": {
- "nullability": "NULLABILITY_NULLABLE"
- }
- },
- {
- "date": {
- "nullability": "NULLABILITY_NULLABLE"
- }
+ "input": {
+ "read": {
+ "common": {
+ "direct": {}
+ },
+ "baseSchema": {
+ "names": [
+ "l_orderkey",
+ "l_partkey",
+ "l_suppkey",
+ "l_linenumber",
+ "l_quantity",
+ "l_extendedprice",
+ "l_discount",
+ "l_tax",
+ "l_returnflag",
+ "l_linestatus",
+ "l_shipdate",
+ "l_commitdate",
+ "l_receiptdate",
+ "l_shipinstruct",
+ "l_shipmode",
+ "l_comment"
+ ],
+ "struct": {
+ "types": [
+ {
+ "i64": {
+ "nullability": "NULLABILITY_NULLABLE"
+ }
+ },
+ {
+ "i64": {
+ "nullability": "NULLABILITY_NULLABLE"
+ }
+ },
+ {
+ "i64": {
+ "nullability": "NULLABILITY_NULLABLE"
+ }
+ },
+ {
+ "i64": {
+ "nullability": "NULLABILITY_NULLABLE"
+ }
+ },
+ {
+ "fp64": {
+ "nullability": "NULLABILITY_NULLABLE"
+ }
+ },
+ {
+ "fp64": {
+ "nullability": "NULLABILITY_NULLABLE"
+ }
+ },
+ {
+ "fp64": {
+ "nullability": "NULLABILITY_NULLABLE"
+ }
+ },
+ {
+ "fp64": {
+ "nullability": "NULLABILITY_NULLABLE"
+ }
+ },
+ {
+ "string": {
+ "nullability": "NULLABILITY_NULLABLE"
+ }
+ },
+ {
+ "string": {
+ "nullability": "NULLABILITY_NULLABLE"
+ }
+ },
+ {
+ "date": {
+ "nullability": "NULLABILITY_NULLABLE"
+ }
+ },
+ {
+ "date": {
+ "nullability": "NULLABILITY_NULLABLE"
+ }
+ },
+ {
+ "date": {
+ "nullability": "NULLABILITY_NULLABLE"
+ }
+ },
+ {
+ "string": {
+ "nullability": "NULLABILITY_NULLABLE"
+ }
+ },
+ {
+ "string": {
+ "nullability": "NULLABILITY_NULLABLE"
+ }
+ },
+ {
+ "string": {
+ "nullability": "NULLABILITY_NULLABLE"
+ }
+ }
+ ]
},
- {
- "date": {
- "nullability": "NULLABILITY_NULLABLE"
- }
- },
- {
- "date": {
- "nullability": "NULLABILITY_NULLABLE"
- }
- },
- {
- "string": {
- "nullability": "NULLABILITY_NULLABLE"
- }
- },
- {
- "string": {
- "nullability": "NULLABILITY_NULLABLE"
- }
- },
- {
- "string": {
- "nullability": "NULLABILITY_NULLABLE"
+ "columnTypes": [
+ "NORMAL_COL",
+ "NORMAL_COL",
+ "NORMAL_COL",
+ "NORMAL_COL",
+ "NORMAL_COL",
+ "NORMAL_COL",
+ "NORMAL_COL",
+ "NORMAL_COL",
+ "NORMAL_COL",
+ "NORMAL_COL",
+ "NORMAL_COL",
+ "NORMAL_COL",
+ "NORMAL_COL",
+ "NORMAL_COL",
+ "NORMAL_COL",
+ "NORMAL_COL"
+ ]
+ },
+ "advancedExtension": {
+ "optimization": {
+ "@type": "type.googleapis.com/google.protobuf.StringValue",
+ "value": "isMergeTree=0\n"
+ }
+ }
+ }
+ },
+ "sorts": [
+ {
+ "expr": {
+ "selection": {
+ "directReference": {
+ "structField": {
+ "field": 10
+ }
}
}
- ]
+ },
+ "direction": "SORT_DIRECTION_ASC_NULLS_FIRST"
},
- "columnTypes": [
- "NORMAL_COL",
- "NORMAL_COL",
- "NORMAL_COL",
- "NORMAL_COL",
- "NORMAL_COL",
- "NORMAL_COL",
- "NORMAL_COL",
- "NORMAL_COL",
- "NORMAL_COL",
- "NORMAL_COL",
- "NORMAL_COL",
- "NORMAL_COL",
- "NORMAL_COL",
- "NORMAL_COL",
- "NORMAL_COL",
- "NORMAL_COL"
- ]
- },
- "advancedExtension": {
- "optimization": {
- "@type": "type.googleapis.com/google.protobuf.StringValue",
- "value": "isMergeTree=0\n"
+ {
+ "expr": {
+ "selection": {
+ "directReference": {
+ "structField": {
+ "field": 8
+ }
+ }
+ }
+ },
+ "direction": "SORT_DIRECTION_ASC_NULLS_FIRST"
}
- }
+ ]
}
}
}
diff --git a/gluten-substrait/src/main/scala/org/apache/gluten/backendsapi/TransformerApi.scala b/gluten-substrait/src/main/scala/org/apache/gluten/backendsapi/TransformerApi.scala
index 69cea9c547..984450bf16 100644
--- a/gluten-substrait/src/main/scala/org/apache/gluten/backendsapi/TransformerApi.scala
+++ b/gluten-substrait/src/main/scala/org/apache/gluten/backendsapi/TransformerApi.scala
@@ -16,12 +16,13 @@
*/
package org.apache.gluten.backendsapi
+import org.apache.gluten.execution.WriteFilesExecTransformer
import org.apache.gluten.substrait.expression.ExpressionNode
import org.apache.spark.sql.catalyst.expressions.{Attribute, Expression}
import org.apache.spark.sql.connector.read.InputPartition
import org.apache.spark.sql.execution.SparkPlan
-import org.apache.spark.sql.execution.datasources.{FileFormat, HadoopFsRelation, PartitionDirectory}
+import org.apache.spark.sql.execution.datasources.{HadoopFsRelation, PartitionDirectory}
import org.apache.spark.sql.types.{DataType, DecimalType, StructType}
import org.apache.spark.util.collection.BitSet
@@ -75,7 +76,7 @@ trait TransformerApi {
/** This method is only used for CH backend tests */
def invalidateSQLExecutionResource(executionId: String): Unit = {}
- def genWriteParameters(fileFormat: FileFormat, writeOptions: Map[String, String]): Any
+ def genWriteParameters(write: WriteFilesExecTransformer): Any
/** use Hadoop Path class to encode the file path */
def encodeFilePathIfNeed(filePath: String): String = filePath
diff --git a/gluten-substrait/src/main/scala/org/apache/gluten/execution/WriteFilesExecTransformer.scala b/gluten-substrait/src/main/scala/org/apache/gluten/execution/WriteFilesExecTransformer.scala
index a9d3a6282a..726dbdc3ef 100644
--- a/gluten-substrait/src/main/scala/org/apache/gluten/execution/WriteFilesExecTransformer.scala
+++ b/gluten-substrait/src/main/scala/org/apache/gluten/execution/WriteFilesExecTransformer.scala
@@ -67,7 +67,7 @@ case class WriteFilesExecTransformer(
override def output: Seq[Attribute] = Seq.empty
- private val caseInsensitiveOptions = CaseInsensitiveMap(options)
+ val caseInsensitiveOptions: CaseInsensitiveMap[String] = CaseInsensitiveMap(options)
def getRelNode(
context: SubstraitContext,
@@ -99,8 +99,7 @@ case class WriteFilesExecTransformer(
ConverterUtils.collectAttributeNames(inputAttributes.toSeq)
val extensionNode = if (!validation) {
ExtensionBuilder.makeAdvancedExtension(
- BackendsApiManager.getTransformerApiInstance
- .genWriteParameters(fileFormat, caseInsensitiveOptions),
+ BackendsApiManager.getTransformerApiInstance.genWriteParameters(this),
SubstraitUtil.createEnhancement(originalInputAttributes)
)
} else {
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org