You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@gluten.apache.org by ho...@apache.org on 2026/03/16 12:05:30 UTC
(gluten) branch main updated: [VL] Gluten-it: Vairous fixes (#11768)
This is an automated email from the ASF dual-hosted git repository.
hongze pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/gluten.git
The following commit(s) were added to refs/heads/main by this push:
new 01cfd98639 [VL] Gluten-it: Vairous fixes (#11768)
01cfd98639 is described below
commit 01cfd9863922a949da006473bf92b6da951deb1a
Author: Hongze Zhang <ho...@gmail.com>
AuthorDate: Mon Mar 16 12:05:25 2026 +0000
[VL] Gluten-it: Vairous fixes (#11768)
---
.../apache/gluten/integration/QueryRunner.scala | 1 +
.../org/apache/gluten/integration/Suite.scala | 1 +
.../apache/gluten/integration/TableCreator.scala | 94 ----
.../gluten/integration/action/Parameterized.scala | 6 -
.../apache/gluten/integration/action/Queries.scala | 13 +-
.../gluten/integration/action/QueriesCompare.scala | 3 -
.../apache/gluten/integration/action/package.scala | 1 -
.../integration/clickbench/ClickBenchSuite.scala | 3 +-
.../clickbench/ClickBenchTableCreator.scala | 2 +-
.../gluten/integration/ds/TpcdsDataGen.scala | 615 +--------------------
.../apache/gluten/integration/ds/TpcdsSuite.scala | 14 +-
.../{TpcdsDataGen.scala => TpcdsTableLayout.scala} | 174 ++----
.../apache/gluten/integration/h/TpchSuite.scala | 3 +-
.../gluten/integration/metrics/PlanMetric.scala | 25 +-
.../integration/table/AutoTableCreator.scala | 83 +++
.../integration/table/LayoutTableCreator.scala | 99 ++++
.../integration/{ => table}/TableAnalyzer.scala | 2 +-
.../gluten/integration/table/TableCreator.scala | 33 ++
.../gluten/integration/table/TableLayout.scala | 25 +
.../org/apache/spark/sql/SparkQueryRunner.scala | 71 ++-
20 files changed, 382 insertions(+), 886 deletions(-)
diff --git a/tools/gluten-it/common/src/main/scala/org/apache/gluten/integration/QueryRunner.scala b/tools/gluten-it/common/src/main/scala/org/apache/gluten/integration/QueryRunner.scala
index efbd6dbd7c..bc0cac02c0 100644
--- a/tools/gluten-it/common/src/main/scala/org/apache/gluten/integration/QueryRunner.scala
+++ b/tools/gluten-it/common/src/main/scala/org/apache/gluten/integration/QueryRunner.scala
@@ -17,6 +17,7 @@
package org.apache.gluten.integration
import org.apache.gluten.integration.metrics.MetricMapper
+import org.apache.gluten.integration.table.{TableAnalyzer, TableCreator}
import org.apache.spark.sql.{RunResult, SparkQueryRunner, SparkSession}
diff --git a/tools/gluten-it/common/src/main/scala/org/apache/gluten/integration/Suite.scala b/tools/gluten-it/common/src/main/scala/org/apache/gluten/integration/Suite.scala
index 112321b8d2..eca23473b3 100644
--- a/tools/gluten-it/common/src/main/scala/org/apache/gluten/integration/Suite.scala
+++ b/tools/gluten-it/common/src/main/scala/org/apache/gluten/integration/Suite.scala
@@ -20,6 +20,7 @@ import org.apache.gluten.integration.Constants.TYPE_MODIFIER_DECIMAL_AS_DOUBLE
import org.apache.gluten.integration.action.Action
import org.apache.gluten.integration.metrics.MetricMapper
import org.apache.gluten.integration.report.TestReporter
+import org.apache.gluten.integration.table.{TableAnalyzer, TableCreator}
import org.apache.spark.SparkConf
import org.apache.spark.deploy.history.HistoryServerHelper
diff --git a/tools/gluten-it/common/src/main/scala/org/apache/gluten/integration/TableCreator.scala b/tools/gluten-it/common/src/main/scala/org/apache/gluten/integration/TableCreator.scala
deleted file mode 100644
index 2819645410..0000000000
--- a/tools/gluten-it/common/src/main/scala/org/apache/gluten/integration/TableCreator.scala
+++ /dev/null
@@ -1,94 +0,0 @@
-/*
- * 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.integration
-
-import org.apache.spark.sql.{AnalysisException, SparkSession}
-
-import org.apache.hadoop.conf.Configuration
-import org.apache.hadoop.fs.{FileSystem, Path}
-
-import java.net.URI
-
-import scala.collection.mutable
-import scala.collection.mutable.ArrayBuffer
-
-trait TableCreator {
- def create(spark: SparkSession, source: String, dataPath: String): Unit
-}
-
-object TableCreator {
- def discoverFromFiles(): TableCreator = {
- DiscoverFromFiles
- }
-
- /** Discover tables automatically from a given file system path. */
- private object DiscoverFromFiles extends TableCreator {
- override def create(spark: SparkSession, source: String, dataPath: String): Unit = {
- val uri = URI.create(dataPath)
- val fs = FileSystem.get(uri, spark.sessionState.newHadoopConf())
-
- val basePath = new Path(dataPath)
- val statuses = fs.listStatus(basePath)
-
- val tableDirs = statuses.filter(_.isDirectory).map(_.getPath)
-
- val tableNames = ArrayBuffer[String]()
-
- val existedTableNames = mutable.ArrayBuffer[String]()
- val createdTableNames = mutable.ArrayBuffer[String]()
- val recoveredPartitionTableNames = mutable.ArrayBuffer[String]()
-
- tableDirs.foreach {
- tablePath =>
- val tableName = tablePath.getName
- tableNames += tableName
- }
-
- println("Creating catalog tables: " + tableNames.mkString(", ") + "...")
-
- tableDirs.foreach {
- tablePath =>
- val tableName = tablePath.getName
- if (spark.catalog.tableExists(tableName)) {
- existedTableNames += tableName
- } else {
- spark.catalog.createTable(tableName, tablePath.toString, source)
- createdTableNames += tableName
- try {
- spark.catalog.recoverPartitions(tableName)
- recoveredPartitionTableNames += tableName
- } catch {
- case _: AnalysisException =>
- }
- }
- }
-
- if (tableNames.isEmpty) {
- return
- }
- if (existedTableNames.nonEmpty) {
- println("Tables already exists: " + existedTableNames.mkString(", ") + ".")
- }
- if (createdTableNames.nonEmpty) {
- println("Tables created: " + createdTableNames.mkString(", ") + ".")
- }
- if (recoveredPartitionTableNames.nonEmpty) {
- println("Recovered partition tables: " + recoveredPartitionTableNames.mkString(", ") + ".")
- }
- }
- }
-}
diff --git a/tools/gluten-it/common/src/main/scala/org/apache/gluten/integration/action/Parameterized.scala b/tools/gluten-it/common/src/main/scala/org/apache/gluten/integration/action/Parameterized.scala
index 3368135122..3264a8e799 100644
--- a/tools/gluten-it/common/src/main/scala/org/apache/gluten/integration/action/Parameterized.scala
+++ b/tools/gluten-it/common/src/main/scala/org/apache/gluten/integration/action/Parameterized.scala
@@ -330,11 +330,6 @@ object Parameterized {
coord.queryResult
.asSuccessOption()
.map(_.runResult.executorMetrics(metricName)))))
- coords.foreach(
- coord =>
- inc
- .next()
- .write(coord.queryResult.asSuccessOption().map(_.runResult.planningTimeMillis)))
coords.foreach(
coord =>
inc
@@ -356,7 +351,6 @@ object Parameterized {
Seq(Field.Branch("Succeeded", coordFields), Field.Branch("Row Count", coordFields)) ++
metricNames.map(metricName => Field.Branch(metricName, coordFields)) ++
Seq(
- Field.Branch("Planning Time (Millis)", coordFields),
Field.Branch("Query Time (Millis)", coordFields)
)
val render =
diff --git a/tools/gluten-it/common/src/main/scala/org/apache/gluten/integration/action/Queries.scala b/tools/gluten-it/common/src/main/scala/org/apache/gluten/integration/action/Queries.scala
index c88cdbb32a..d6ba8b3601 100644
--- a/tools/gluten-it/common/src/main/scala/org/apache/gluten/integration/action/Queries.scala
+++ b/tools/gluten-it/common/src/main/scala/org/apache/gluten/integration/action/Queries.scala
@@ -16,12 +16,13 @@
*/
package org.apache.gluten.integration.action
-import org.apache.gluten.integration.{Query, QueryRunner, Suite, TableCreator}
+import org.apache.gluten.integration.{Query, QueryRunner, Suite}
import org.apache.gluten.integration.QueryRunner.QueryResult
import org.apache.gluten.integration.action.Actions.QuerySelector
import org.apache.gluten.integration.action.TableRender.RowParser.FieldAppender.RowAppender
import org.apache.gluten.integration.metrics.{MetricMapper, PlanMetric}
import org.apache.gluten.integration.stat.RamStat
+import org.apache.gluten.integration.table.TableCreator
import org.apache.spark.sql.SparkSession
@@ -150,24 +151,18 @@ object Queries {
line.queryResult match {
case QueryRunner.Success(_, runResult) =>
inc.next().write(runResult.rows.size)
- inc.next().write(runResult.planningTimeMillis)
inc.next().write(runResult.executionTimeMillis)
case QueryRunner.Failure(_, error) =>
inc.next().write(None)
inc.next().write(None)
- inc.next().write(None)
}
}
}
}
private def printResults(out: PrintStream, results: Seq[TestResultLine]): Unit = {
- val render = TableRender.plain[TestResultLine](
- "Query ID",
- "Was Passed",
- "Row Count",
- "Plan Time (Millis)",
- "Query Time (Millis)")
+ val render = TableRender
+ .plain[TestResultLine]("Query ID", "Was Passed", "Row Count", "Query Time (Millis)")
results.foreach(line => render.appendRow(line))
diff --git a/tools/gluten-it/common/src/main/scala/org/apache/gluten/integration/action/QueriesCompare.scala b/tools/gluten-it/common/src/main/scala/org/apache/gluten/integration/action/QueriesCompare.scala
index e5f0c381e9..4c35945c92 100644
--- a/tools/gluten-it/common/src/main/scala/org/apache/gluten/integration/action/QueriesCompare.scala
+++ b/tools/gluten-it/common/src/main/scala/org/apache/gluten/integration/action/QueriesCompare.scala
@@ -196,8 +196,6 @@ object QueriesCompare {
inc.next().write(line.testPassed())
inc.next().write(line.expected.asSuccessOption().map(_.runResult.rows.size))
inc.next().write(line.actual.asSuccessOption().map(_.runResult.rows.size))
- inc.next().write(line.expected.asSuccessOption().map(_.runResult.planningTimeMillis))
- inc.next().write(line.actual.asSuccessOption().map(_.runResult.planningTimeMillis))
inc.next().write(line.expected.asSuccessOption().map(_.runResult.executionTimeMillis))
inc.next().write(line.actual.asSuccessOption().map(_.runResult.executionTimeMillis))
@@ -222,7 +220,6 @@ object QueriesCompare {
Leaf("Query ID"),
Leaf("Passed"),
Branch("Row Count", List(Leaf("Vanilla"), Leaf("Gluten"))),
- Branch("Planning Time (Millis)", List(Leaf("Vanilla"), Leaf("Gluten"))),
Branch("Query Time (Millis)", List(Leaf("Vanilla"), Leaf("Gluten"))),
Leaf("Speedup")
)
diff --git a/tools/gluten-it/common/src/main/scala/org/apache/gluten/integration/action/package.scala b/tools/gluten-it/common/src/main/scala/org/apache/gluten/integration/action/package.scala
index 84a1d296b4..6945e1adce 100644
--- a/tools/gluten-it/common/src/main/scala/org/apache/gluten/integration/action/package.scala
+++ b/tools/gluten-it/common/src/main/scala/org/apache/gluten/integration/action/package.scala
@@ -38,7 +38,6 @@ package object action {
name,
RunResult(
c1.runResult.rows ++ c2.runResult.rows,
- c1.runResult.planningTimeMillis + c2.runResult.planningTimeMillis,
c1.runResult.executionTimeMillis + c2.runResult.executionTimeMillis,
c1.runResult.sqlMetrics ++ c2.runResult.sqlMetrics,
(c1.runResult.executorMetrics, c2.runResult.executorMetrics).sumUp
diff --git a/tools/gluten-it/common/src/main/scala/org/apache/gluten/integration/clickbench/ClickBenchSuite.scala b/tools/gluten-it/common/src/main/scala/org/apache/gluten/integration/clickbench/ClickBenchSuite.scala
index 1eb4c3bd07..a93c7e3746 100644
--- a/tools/gluten-it/common/src/main/scala/org/apache/gluten/integration/clickbench/ClickBenchSuite.scala
+++ b/tools/gluten-it/common/src/main/scala/org/apache/gluten/integration/clickbench/ClickBenchSuite.scala
@@ -16,10 +16,11 @@
*/
package org.apache.gluten.integration.clickbench
-import org.apache.gluten.integration.{DataGen, QuerySet, Suite, TableAnalyzer, TableCreator}
+import org.apache.gluten.integration.{DataGen, QuerySet, Suite}
import org.apache.gluten.integration.action.Action
import org.apache.gluten.integration.metrics.MetricMapper
import org.apache.gluten.integration.report.TestReporter
+import org.apache.gluten.integration.table.{TableAnalyzer, TableCreator}
import org.apache.spark.SparkConf
diff --git a/tools/gluten-it/common/src/main/scala/org/apache/gluten/integration/clickbench/ClickBenchTableCreator.scala b/tools/gluten-it/common/src/main/scala/org/apache/gluten/integration/clickbench/ClickBenchTableCreator.scala
index 4e83ee9f71..3f7140a8fd 100644
--- a/tools/gluten-it/common/src/main/scala/org/apache/gluten/integration/clickbench/ClickBenchTableCreator.scala
+++ b/tools/gluten-it/common/src/main/scala/org/apache/gluten/integration/clickbench/ClickBenchTableCreator.scala
@@ -16,7 +16,7 @@
*/
package org.apache.gluten.integration.clickbench
-import org.apache.gluten.integration.TableCreator
+import org.apache.gluten.integration.table.TableCreator
import org.apache.spark.sql.{AnalysisException, SparkSession}
import org.apache.spark.sql.types.StructType
diff --git a/tools/gluten-it/common/src/main/scala/org/apache/gluten/integration/ds/TpcdsDataGen.scala b/tools/gluten-it/common/src/main/scala/org/apache/gluten/integration/ds/TpcdsDataGen.scala
index b80709edb7..8422081d2f 100644
--- a/tools/gluten-it/common/src/main/scala/org/apache/gluten/integration/ds/TpcdsDataGen.scala
+++ b/tools/gluten-it/common/src/main/scala/org/apache/gluten/integration/ds/TpcdsDataGen.scala
@@ -29,11 +29,11 @@ import java.io.File
import scala.collection.JavaConverters._
class TpcdsDataGen(
+ source: String,
+ tableLayout: TpcdsTableLayout,
scale: Double,
partitions: Int,
- source: String,
dir: String,
- genPartitionedData: Boolean,
featureNames: Seq[String],
typeModifiers: Seq[TypeModifier]
) extends Serializable
@@ -51,42 +51,8 @@ class TpcdsDataGen(
if (name.equals("dbgen_version")) {
return
}
- val schema = name match {
- case "catalog_sales" => TpcdsDataGen.catalogSalesSchema
- case "catalog_returns" => TpcdsDataGen.catalogReturnsSchema
- case "inventory" => TpcdsDataGen.inventorySchema
- case "store_sales" => TpcdsDataGen.storeSalesSchema
- case "store_returns" => TpcdsDataGen.storeReturnsSchema
- case "web_sales" => TpcdsDataGen.webSalesSchema
- case "web_returns" => TpcdsDataGen.webReturnsSchema
- case "call_center" => TpcdsDataGen.callCenterSchema
- case "catalog_page" => TpcdsDataGen.catalogPageSchema
- case "customer" => TpcdsDataGen.customerSchema
- case "customer_address" => TpcdsDataGen.customerAddressSchema
- case "customer_demographics" => TpcdsDataGen.customerDemographicsSchema
- case "date_dim" => TpcdsDataGen.dateDimSchema
- case "household_demographics" => TpcdsDataGen.householdDemographicsSchema
- case "income_band" => TpcdsDataGen.incomeBandSchema
- case "item" => TpcdsDataGen.itemSchema
- case "promotion" => TpcdsDataGen.promotionSchema
- case "reason" => TpcdsDataGen.reasonSchema
- case "ship_mode" => TpcdsDataGen.shipModeSchema
- case "store" => TpcdsDataGen.storeSchema
- case "time_dim" => TpcdsDataGen.timeDimSchema
- case "warehouse" => TpcdsDataGen.warehouseSchema
- case "web_page" => TpcdsDataGen.webPageSchema
- case "web_site" => TpcdsDataGen.webSiteSchema
- }
- val partitionBy: List[String] = if (!genPartitionedData) {
- List[String]()
- } else {
- name match {
- case "store_sales" => List("ss_sold_date_sk")
- case "catalog_sales" => List("cs_sold_date_sk")
- case "web_sales" => List("ws_sold_date_sk")
- case _ => List[String]()
- }
- }
+ val schema = tableLayout.getSchema(name)
+ val partitionBy = tableLayout.getPartitionColumns(name)
writeParquetTable(spark, name, t, schema, partitionBy)
}
@@ -96,7 +62,7 @@ class TpcdsDataGen(
tableName: String,
t: Table,
schema: StructType,
- partitionBy: List[String]): Unit = {
+ partitionBy: Seq[String]): Unit = {
println(s"Generating table $tableName...")
val rowModifier = DataGen.getRowModifier(schema, typeModifiers)
val modifiedSchema = DataGen.modifySchema(schema, rowModifier)
@@ -142,573 +108,4 @@ class TpcdsDataGen(
}
}
-object TpcdsDataGen {
-
- // generated by script
- private def catalogSalesSchema = {
- StructType(
- Seq(
- StructField("cs_sold_date_sk", LongType),
- StructField("cs_sold_time_sk", LongType),
- StructField("cs_ship_date_sk", LongType),
- StructField("cs_bill_customer_sk", LongType),
- StructField("cs_bill_cdemo_sk", LongType),
- StructField("cs_bill_hdemo_sk", LongType),
- StructField("cs_bill_addr_sk", LongType),
- StructField("cs_ship_customer_sk", LongType),
- StructField("cs_ship_cdemo_sk", LongType),
- StructField("cs_ship_hdemo_sk", LongType),
- StructField("cs_ship_addr_sk", LongType),
- StructField("cs_call_center_sk", LongType),
- StructField("cs_catalog_page_sk", LongType),
- StructField("cs_ship_mode_sk", LongType),
- StructField("cs_warehouse_sk", LongType),
- StructField("cs_item_sk", LongType),
- StructField("cs_promo_sk", LongType),
- StructField("cs_order_number", LongType),
- StructField("cs_quantity", LongType),
- StructField("cs_wholesale_cost", DecimalType(7, 2)),
- StructField("cs_list_price", DecimalType(7, 2)),
- StructField("cs_sales_price", DecimalType(7, 2)),
- StructField("cs_ext_discount_amt", DecimalType(7, 2)),
- StructField("cs_ext_sales_price", DecimalType(7, 2)),
- StructField("cs_ext_wholesale_cost", DecimalType(7, 2)),
- StructField("cs_ext_list_price", DecimalType(7, 2)),
- StructField("cs_ext_tax", DecimalType(7, 2)),
- StructField("cs_coupon_amt", DecimalType(7, 2)),
- StructField("cs_ext_ship_cost", DecimalType(7, 2)),
- StructField("cs_net_paid", DecimalType(7, 2)),
- StructField("cs_net_paid_inc_tax", DecimalType(7, 2)),
- StructField("cs_net_paid_inc_ship", DecimalType(7, 2)),
- StructField("cs_net_paid_inc_ship_tax", DecimalType(7, 2)),
- StructField("cs_net_profit", DecimalType(7, 2))
- ))
- }
-
- private def catalogReturnsSchema = {
- StructType(
- Seq(
- StructField("cr_returned_date_sk", LongType),
- StructField("cr_returned_time_sk", LongType),
- StructField("cr_item_sk", LongType),
- StructField("cr_refunded_customer_sk", LongType),
- StructField("cr_refunded_cdemo_sk", LongType),
- StructField("cr_refunded_hdemo_sk", LongType),
- StructField("cr_refunded_addr_sk", LongType),
- StructField("cr_returning_customer_sk", LongType),
- StructField("cr_returning_cdemo_sk", LongType),
- StructField("cr_returning_hdemo_sk", LongType),
- StructField("cr_returning_addr_sk", LongType),
- StructField("cr_call_center_sk", LongType),
- StructField("cr_catalog_page_sk", LongType),
- StructField("cr_ship_mode_sk", LongType),
- StructField("cr_warehouse_sk", LongType),
- StructField("cr_reason_sk", LongType),
- StructField("cr_order_number", LongType),
- StructField("cr_return_quantity", LongType),
- StructField("cr_return_amount", DecimalType(7, 2)),
- StructField("cr_return_tax", DecimalType(7, 2)),
- StructField("cr_return_amt_inc_tax", DecimalType(7, 2)),
- StructField("cr_fee", DecimalType(7, 2)),
- StructField("cr_return_ship_cost", DecimalType(7, 2)),
- StructField("cr_refunded_cash", DecimalType(7, 2)),
- StructField("cr_reversed_charge", DecimalType(7, 2)),
- StructField("cr_store_credit", DecimalType(7, 2)),
- StructField("cr_net_loss", DecimalType(7, 2))
- ))
- }
-
- private def inventorySchema = {
- StructType(
- Seq(
- StructField("inv_date_sk", LongType),
- StructField("inv_item_sk", LongType),
- StructField("inv_warehouse_sk", LongType),
- StructField("inv_quantity_on_hand", LongType)
- ))
- }
-
- private def storeSalesSchema = {
- StructType(
- Seq(
- StructField("ss_sold_date_sk", LongType),
- StructField("ss_sold_time_sk", LongType),
- StructField("ss_item_sk", LongType),
- StructField("ss_customer_sk", LongType),
- StructField("ss_cdemo_sk", LongType),
- StructField("ss_hdemo_sk", LongType),
- StructField("ss_addr_sk", LongType),
- StructField("ss_store_sk", LongType),
- StructField("ss_promo_sk", LongType),
- StructField("ss_ticket_number", LongType),
- StructField("ss_quantity", LongType),
- StructField("ss_wholesale_cost", DecimalType(7, 2)),
- StructField("ss_list_price", DecimalType(7, 2)),
- StructField("ss_sales_price", DecimalType(7, 2)),
- StructField("ss_ext_discount_amt", DecimalType(7, 2)),
- StructField("ss_ext_sales_price", DecimalType(7, 2)),
- StructField("ss_ext_wholesale_cost", DecimalType(7, 2)),
- StructField("ss_ext_list_price", DecimalType(7, 2)),
- StructField("ss_ext_tax", DecimalType(7, 2)),
- StructField("ss_coupon_amt", DecimalType(7, 2)),
- StructField("ss_net_paid", DecimalType(7, 2)),
- StructField("ss_net_paid_inc_tax", DecimalType(7, 2)),
- StructField("ss_net_profit", DecimalType(7, 2))
- ))
- }
-
- private def storeReturnsSchema = {
- StructType(
- Seq(
- StructField("sr_returned_date_sk", LongType),
- StructField("sr_return_time_sk", LongType),
- StructField("sr_item_sk", LongType),
- StructField("sr_customer_sk", LongType),
- StructField("sr_cdemo_sk", LongType),
- StructField("sr_hdemo_sk", LongType),
- StructField("sr_addr_sk", LongType),
- StructField("sr_store_sk", LongType),
- StructField("sr_reason_sk", LongType),
- StructField("sr_ticket_number", LongType),
- StructField("sr_return_quantity", LongType),
- StructField("sr_return_amt", DecimalType(7, 2)),
- StructField("sr_return_tax", DecimalType(7, 2)),
- StructField("sr_return_amt_inc_tax", DecimalType(7, 2)),
- StructField("sr_fee", DecimalType(7, 2)),
- StructField("sr_return_ship_cost", DecimalType(7, 2)),
- StructField("sr_refunded_cash", DecimalType(7, 2)),
- StructField("sr_reversed_charge", DecimalType(7, 2)),
- StructField("sr_store_credit", DecimalType(7, 2)),
- StructField("sr_net_loss", DecimalType(7, 2))
- ))
- }
-
- private def webSalesSchema = {
- StructType(
- Seq(
- StructField("ws_sold_date_sk", LongType),
- StructField("ws_sold_time_sk", LongType),
- StructField("ws_ship_date_sk", LongType),
- StructField("ws_item_sk", LongType),
- StructField("ws_bill_customer_sk", LongType),
- StructField("ws_bill_cdemo_sk", LongType),
- StructField("ws_bill_hdemo_sk", LongType),
- StructField("ws_bill_addr_sk", LongType),
- StructField("ws_ship_customer_sk", LongType),
- StructField("ws_ship_cdemo_sk", LongType),
- StructField("ws_ship_hdemo_sk", LongType),
- StructField("ws_ship_addr_sk", LongType),
- StructField("ws_web_page_sk", LongType),
- StructField("ws_web_site_sk", LongType),
- StructField("ws_ship_mode_sk", LongType),
- StructField("ws_warehouse_sk", LongType),
- StructField("ws_promo_sk", LongType),
- StructField("ws_order_number", LongType),
- StructField("ws_quantity", LongType),
- StructField("ws_wholesale_cost", DecimalType(7, 2)),
- StructField("ws_list_price", DecimalType(7, 2)),
- StructField("ws_sales_price", DecimalType(7, 2)),
- StructField("ws_ext_discount_amt", DecimalType(7, 2)),
- StructField("ws_ext_sales_price", DecimalType(7, 2)),
- StructField("ws_ext_wholesale_cost", DecimalType(7, 2)),
- StructField("ws_ext_list_price", DecimalType(7, 2)),
- StructField("ws_ext_tax", DecimalType(7, 2)),
- StructField("ws_coupon_amt", DecimalType(7, 2)),
- StructField("ws_ext_ship_cost", DecimalType(7, 2)),
- StructField("ws_net_paid", DecimalType(7, 2)),
- StructField("ws_net_paid_inc_tax", DecimalType(7, 2)),
- StructField("ws_net_paid_inc_ship", DecimalType(7, 2)),
- StructField("ws_net_paid_inc_ship_tax", DecimalType(7, 2)),
- StructField("ws_net_profit", DecimalType(7, 2))
- ))
- }
-
- private def webReturnsSchema = {
- StructType(
- Seq(
- StructField("wr_returned_date_sk", LongType),
- StructField("wr_returned_time_sk", LongType),
- StructField("wr_item_sk", LongType),
- StructField("wr_refunded_customer_sk", LongType),
- StructField("wr_refunded_cdemo_sk", LongType),
- StructField("wr_refunded_hdemo_sk", LongType),
- StructField("wr_refunded_addr_sk", LongType),
- StructField("wr_returning_customer_sk", LongType),
- StructField("wr_returning_cdemo_sk", LongType),
- StructField("wr_returning_hdemo_sk", LongType),
- StructField("wr_returning_addr_sk", LongType),
- StructField("wr_web_page_sk", LongType),
- StructField("wr_reason_sk", LongType),
- StructField("wr_order_number", LongType),
- StructField("wr_return_quantity", LongType),
- StructField("wr_return_amt", DecimalType(7, 2)),
- StructField("wr_return_tax", DecimalType(7, 2)),
- StructField("wr_return_amt_inc_tax", DecimalType(7, 2)),
- StructField("wr_fee", DecimalType(7, 2)),
- StructField("wr_return_ship_cost", DecimalType(7, 2)),
- StructField("wr_refunded_cash", DecimalType(7, 2)),
- StructField("wr_reversed_charge", DecimalType(7, 2)),
- StructField("wr_account_credit", DecimalType(7, 2)),
- StructField("wr_net_loss", DecimalType(7, 2))
- ))
- }
-
- private def callCenterSchema = {
- StructType(
- Seq(
- StructField("cc_call_center_sk", LongType),
- StructField("cc_call_center_id", StringType),
- StructField("cc_rec_start_date", DateType),
- StructField("cc_rec_end_date", DateType),
- StructField("cc_closed_date_sk", LongType),
- StructField("cc_open_date_sk", LongType),
- StructField("cc_name", StringType),
- StructField("cc_class", StringType),
- StructField("cc_employees", LongType),
- StructField("cc_sq_ft", LongType),
- StructField("cc_hours", StringType),
- StructField("cc_manager", StringType),
- StructField("cc_mkt_id", LongType),
- StructField("cc_mkt_class", StringType),
- StructField("cc_mkt_desc", StringType),
- StructField("cc_market_manager", StringType),
- StructField("cc_division", LongType),
- StructField("cc_division_name", StringType),
- StructField("cc_company", LongType),
- StructField("cc_company_name", StringType),
- StructField("cc_street_number", StringType),
- StructField("cc_street_name", StringType),
- StructField("cc_street_type", StringType),
- StructField("cc_suite_number", StringType),
- StructField("cc_city", StringType),
- StructField("cc_county", StringType),
- StructField("cc_state", StringType),
- StructField("cc_zip", StringType),
- StructField("cc_country", StringType),
- StructField("cc_gmt_offset", DecimalType(5, 2)),
- StructField("cc_tax_percentage", DecimalType(5, 2))
- ))
- }
-
- private def catalogPageSchema = {
- StructType(
- Seq(
- StructField("cp_catalog_page_sk", LongType),
- StructField("cp_catalog_page_id", StringType),
- StructField("cp_start_date_sk", LongType),
- StructField("cp_end_date_sk", LongType),
- StructField("cp_department", StringType),
- StructField("cp_catalog_number", LongType),
- StructField("cp_catalog_page_number", LongType),
- StructField("cp_description", StringType),
- StructField("cp_type", StringType)
- ))
- }
-
- private def customerSchema = {
- StructType(
- Seq(
- StructField("c_customer_sk", LongType),
- StructField("c_customer_id", StringType),
- StructField("c_current_cdemo_sk", LongType),
- StructField("c_current_hdemo_sk", LongType),
- StructField("c_current_addr_sk", LongType),
- StructField("c_first_shipto_date_sk", LongType),
- StructField("c_first_sales_date_sk", LongType),
- StructField("c_salutation", StringType),
- StructField("c_first_name", StringType),
- StructField("c_last_name", StringType),
- StructField("c_preferred_cust_flag", StringType),
- StructField("c_birth_day", LongType),
- StructField("c_birth_month", LongType),
- StructField("c_birth_year", LongType),
- StructField("c_birth_country", StringType),
- StructField("c_login", StringType),
- StructField("c_email_address", StringType),
- StructField("c_last_review_date", StringType)
- ))
- }
-
- private def customerAddressSchema = {
- StructType(
- Seq(
- StructField("ca_address_sk", LongType),
- StructField("ca_address_id", StringType),
- StructField("ca_street_number", StringType),
- StructField("ca_street_name", StringType),
- StructField("ca_street_type", StringType),
- StructField("ca_suite_number", StringType),
- StructField("ca_city", StringType),
- StructField("ca_county", StringType),
- StructField("ca_state", StringType),
- StructField("ca_zip", StringType),
- StructField("ca_country", StringType),
- StructField("ca_gmt_offset", DecimalType(5, 2)),
- StructField("ca_location_type", StringType)
- ))
- }
-
- private def customerDemographicsSchema = {
- StructType(
- Seq(
- StructField("cd_demo_sk", LongType),
- StructField("cd_gender", StringType),
- StructField("cd_marital_status", StringType),
- StructField("cd_education_status", StringType),
- StructField("cd_purchase_estimate", LongType),
- StructField("cd_credit_rating", StringType),
- StructField("cd_dep_count", LongType),
- StructField("cd_dep_employed_count", LongType),
- StructField("cd_dep_college_count", LongType)
- ))
- }
-
- private def dateDimSchema = {
- StructType(
- Seq(
- StructField("d_date_sk", LongType),
- StructField("d_date_id", StringType),
- StructField("d_date", DateType),
- StructField("d_month_seq", LongType),
- StructField("d_week_seq", LongType),
- StructField("d_quarter_seq", LongType),
- StructField("d_year", LongType),
- StructField("d_dow", LongType),
- StructField("d_moy", LongType),
- StructField("d_dom", LongType),
- StructField("d_qoy", LongType),
- StructField("d_fy_year", LongType),
- StructField("d_fy_quarter_seq", LongType),
- StructField("d_fy_week_seq", LongType),
- StructField("d_day_name", StringType),
- StructField("d_quarter_name", StringType),
- StructField("d_holiday", StringType),
- StructField("d_weekend", StringType),
- StructField("d_following_holiday", StringType),
- StructField("d_first_dom", LongType),
- StructField("d_last_dom", LongType),
- StructField("d_same_day_ly", LongType),
- StructField("d_same_day_lq", LongType),
- StructField("d_current_day", StringType),
- StructField("d_current_week", StringType),
- StructField("d_current_month", StringType),
- StructField("d_current_quarter", StringType),
- StructField("d_current_year", StringType)
- ))
- }
-
- private def householdDemographicsSchema = {
- StructType(
- Seq(
- StructField("hd_demo_sk", LongType),
- StructField("hd_income_band_sk", LongType),
- StructField("hd_buy_potential", StringType),
- StructField("hd_dep_count", LongType),
- StructField("hd_vehicle_count", LongType)
- ))
- }
-
- private def incomeBandSchema = {
- StructType(
- Seq(
- StructField("ib_income_band_sk", LongType),
- StructField("ib_lower_bound", LongType),
- StructField("ib_upper_bound", LongType)))
- }
-
- private def itemSchema = {
- StructType(
- Seq(
- StructField("i_item_sk", LongType),
- StructField("i_item_id", StringType),
- StructField("i_rec_start_date", DateType),
- StructField("i_rec_end_date", DateType),
- StructField("i_item_desc", StringType),
- StructField("i_current_price", DecimalType(7, 2)),
- StructField("i_wholesale_cost", DecimalType(7, 2)),
- StructField("i_brand_id", LongType),
- StructField("i_brand", StringType),
- StructField("i_class_id", LongType),
- StructField("i_class", StringType),
- StructField("i_category_id", LongType),
- StructField("i_category", StringType),
- StructField("i_manufact_id", LongType),
- StructField("i_manufact", StringType),
- StructField("i_size", StringType),
- StructField("i_formulation", StringType),
- StructField("i_color", StringType),
- StructField("i_units", StringType),
- StructField("i_container", StringType),
- StructField("i_manager_id", LongType),
- StructField("i_product_name", StringType)
- ))
- }
-
- private def promotionSchema = {
- StructType(
- Seq(
- StructField("p_promo_sk", LongType),
- StructField("p_promo_id", StringType),
- StructField("p_start_date_sk", LongType),
- StructField("p_end_date_sk", LongType),
- StructField("p_item_sk", LongType),
- StructField("p_cost", DecimalType(15, 2)),
- StructField("p_response_target", LongType),
- StructField("p_promo_name", StringType),
- StructField("p_channel_dmail", StringType),
- StructField("p_channel_email", StringType),
- StructField("p_channel_catalog", StringType),
- StructField("p_channel_tv", StringType),
- StructField("p_channel_radio", StringType),
- StructField("p_channel_press", StringType),
- StructField("p_channel_event", StringType),
- StructField("p_channel_demo", StringType),
- StructField("p_channel_details", StringType),
- StructField("p_purpose", StringType),
- StructField("p_discount_active", StringType)
- ))
- }
-
- private def reasonSchema = {
- StructType(
- Seq(
- StructField("r_reason_sk", LongType),
- StructField("r_reason_id", StringType),
- StructField("r_reason_desc", StringType)))
- }
-
- private def shipModeSchema = {
- StructType(
- Seq(
- StructField("sm_ship_mode_sk", LongType),
- StructField("sm_ship_mode_id", StringType),
- StructField("sm_type", StringType),
- StructField("sm_code", StringType),
- StructField("sm_carrier", StringType),
- StructField("sm_contract", StringType)
- ))
- }
-
- private def storeSchema = {
- StructType(
- Seq(
- StructField("s_store_sk", LongType),
- StructField("s_store_id", StringType),
- StructField("s_rec_start_date", DateType),
- StructField("s_rec_end_date", DateType),
- StructField("s_closed_date_sk", LongType),
- StructField("s_store_name", StringType),
- StructField("s_number_employees", LongType),
- StructField("s_floor_space", LongType),
- StructField("s_hours", StringType),
- StructField("s_manager", StringType),
- StructField("s_market_id", LongType),
- StructField("s_geography_class", StringType),
- StructField("s_market_desc", StringType),
- StructField("s_market_manager", StringType),
- StructField("s_division_id", LongType),
- StructField("s_division_name", StringType),
- StructField("s_company_id", LongType),
- StructField("s_company_name", StringType),
- StructField("s_street_number", StringType),
- StructField("s_street_name", StringType),
- StructField("s_street_type", StringType),
- StructField("s_suite_number", StringType),
- StructField("s_city", StringType),
- StructField("s_county", StringType),
- StructField("s_state", StringType),
- StructField("s_zip", StringType),
- StructField("s_country", StringType),
- StructField("s_gmt_offset", DecimalType(5, 2)),
- StructField("s_tax_precentage", DecimalType(5, 2))
- ))
- }
-
- private def timeDimSchema = {
- StructType(
- Seq(
- StructField("t_time_sk", LongType),
- StructField("t_time_id", StringType),
- StructField("t_time", LongType),
- StructField("t_hour", LongType),
- StructField("t_minute", LongType),
- StructField("t_second", LongType),
- StructField("t_am_pm", StringType),
- StructField("t_shift", StringType),
- StructField("t_sub_shift", StringType),
- StructField("t_meal_time", StringType)
- ))
- }
-
- private def warehouseSchema = {
- StructType(
- Seq(
- StructField("w_warehouse_sk", LongType),
- StructField("w_warehouse_id", StringType),
- StructField("w_warehouse_name", StringType),
- StructField("w_warehouse_sq_ft", LongType),
- StructField("w_street_number", StringType),
- StructField("w_street_name", StringType),
- StructField("w_street_type", StringType),
- StructField("w_suite_number", StringType),
- StructField("w_city", StringType),
- StructField("w_county", StringType),
- StructField("w_state", StringType),
- StructField("w_zip", StringType),
- StructField("w_country", StringType),
- StructField("w_gmt_offset", DecimalType(5, 2))
- ))
- }
-
- private def webPageSchema = {
- StructType(
- Seq(
- StructField("wp_web_page_sk", LongType),
- StructField("wp_web_page_id", StringType),
- StructField("wp_rec_start_date", DateType),
- StructField("wp_rec_end_date", DateType),
- StructField("wp_creation_date_sk", LongType),
- StructField("wp_access_date_sk", LongType),
- StructField("wp_autogen_flag", StringType),
- StructField("wp_customer_sk", LongType),
- StructField("wp_url", StringType),
- StructField("wp_type", StringType),
- StructField("wp_char_count", LongType),
- StructField("wp_link_count", LongType),
- StructField("wp_image_count", LongType),
- StructField("wp_max_ad_count", LongType)
- ))
- }
-
- private def webSiteSchema = {
- StructType(
- Seq(
- StructField("web_site_sk", LongType),
- StructField("web_site_id", StringType),
- StructField("web_rec_start_date", DateType),
- StructField("web_rec_end_date", DateType),
- StructField("web_name", StringType),
- StructField("web_open_date_sk", LongType),
- StructField("web_close_date_sk", LongType),
- StructField("web_class", StringType),
- StructField("web_manager", StringType),
- StructField("web_mkt_id", LongType),
- StructField("web_mkt_class", StringType),
- StructField("web_mkt_desc", StringType),
- StructField("web_market_manager", StringType),
- StructField("web_company_id", LongType),
- StructField("web_company_name", StringType),
- StructField("web_street_number", StringType),
- StructField("web_street_name", StringType),
- StructField("web_street_type", StringType),
- StructField("web_suite_number", StringType),
- StructField("web_city", StringType),
- StructField("web_county", StringType),
- StructField("web_state", StringType),
- StructField("web_zip", StringType),
- StructField("web_country", StringType),
- StructField("web_gmt_offset", StringType),
- StructField("web_tax_percentage", DecimalType(5, 2))
- ))
- }
-}
+object TpcdsDataGen {}
diff --git a/tools/gluten-it/common/src/main/scala/org/apache/gluten/integration/ds/TpcdsSuite.scala b/tools/gluten-it/common/src/main/scala/org/apache/gluten/integration/ds/TpcdsSuite.scala
index 427c088be2..b99104ded5 100644
--- a/tools/gluten-it/common/src/main/scala/org/apache/gluten/integration/ds/TpcdsSuite.scala
+++ b/tools/gluten-it/common/src/main/scala/org/apache/gluten/integration/ds/TpcdsSuite.scala
@@ -16,9 +16,10 @@
*/
package org.apache.gluten.integration.ds
-import org.apache.gluten.integration.{DataGen, QuerySet, Suite, TableAnalyzer, TableCreator}
+import org.apache.gluten.integration.{DataGen, QuerySet, Suite}
import org.apache.gluten.integration.action.Action
import org.apache.gluten.integration.metrics.MetricMapper
+import org.apache.gluten.integration.table.{TableAnalyzer, TableCreator}
import org.apache.spark.SparkConf
@@ -77,6 +78,10 @@ class TpcdsSuite(
) {
import TpcdsSuite._
+ checkDataGenArgs(dataSource, dataScale, genPartitionedData)
+
+ private val tableLayout = new TpcdsTableLayout(genPartitionedData)
+
override protected def historyWritePath(): String = HISTORY_WRITE_PATH
override private[integration] def dataWritePath(): String = {
@@ -92,13 +97,12 @@ class TpcdsSuite(
}
override private[integration] def createDataGen(): DataGen = {
- checkDataGenArgs(dataSource, dataScale, genPartitionedData)
new TpcdsDataGen(
+ dataSource,
+ tableLayout,
dataScale,
shufflePartitions,
- dataSource,
dataWritePath(),
- genPartitionedData,
dataGenFeatures,
typeModifiers())
}
@@ -109,7 +113,7 @@ class TpcdsSuite(
override private[integration] def desc(): String = "TPC-DS"
- override def tableCreator(): TableCreator = TableCreator.discoverFromFiles()
+ override def tableCreator(): TableCreator = TableCreator.createFromLayout(tableLayout)
override def tableAnalyzer0(): TableAnalyzer = TableAnalyzer.analyzeAll()
}
diff --git a/tools/gluten-it/common/src/main/scala/org/apache/gluten/integration/ds/TpcdsDataGen.scala b/tools/gluten-it/common/src/main/scala/org/apache/gluten/integration/ds/TpcdsTableLayout.scala
similarity index 83%
copy from tools/gluten-it/common/src/main/scala/org/apache/gluten/integration/ds/TpcdsDataGen.scala
copy to tools/gluten-it/common/src/main/scala/org/apache/gluten/integration/ds/TpcdsTableLayout.scala
index b80709edb7..23fb841583 100644
--- a/tools/gluten-it/common/src/main/scala/org/apache/gluten/integration/ds/TpcdsDataGen.scala
+++ b/tools/gluten-it/common/src/main/scala/org/apache/gluten/integration/ds/TpcdsTableLayout.scala
@@ -16,133 +16,67 @@
*/
package org.apache.gluten.integration.ds
-import org.apache.gluten.integration.{DataGen, TypeModifier}
-import org.apache.gluten.integration.shim.Shim
-
-import org.apache.spark.sql.{Column, Row, SaveMode, SparkSession}
-import org.apache.spark.sql.types._
-
-import io.trino.tpcds._
-
-import java.io.File
-
-import scala.collection.JavaConverters._
-
-class TpcdsDataGen(
- scale: Double,
- partitions: Int,
- source: String,
- dir: String,
- genPartitionedData: Boolean,
- featureNames: Seq[String],
- typeModifiers: Seq[TypeModifier]
-) extends Serializable
- with DataGen {
-
- private val featureRegistry = new DataGen.FeatureRegistry
-
- featureRegistry.register(TpcdsDataGenFeatures.EnableDeltaDeletionVector)
- featureRegistry.register(TpcdsDataGenFeatures.DeleteTenPercentData)
-
- private val features = featureNames.map(featureRegistry.getFeature)
-
- def writeParquetTable(spark: SparkSession, t: Table): Unit = {
- val name = t.getName
- if (name.equals("dbgen_version")) {
- return
- }
- val schema = name match {
- case "catalog_sales" => TpcdsDataGen.catalogSalesSchema
- case "catalog_returns" => TpcdsDataGen.catalogReturnsSchema
- case "inventory" => TpcdsDataGen.inventorySchema
- case "store_sales" => TpcdsDataGen.storeSalesSchema
- case "store_returns" => TpcdsDataGen.storeReturnsSchema
- case "web_sales" => TpcdsDataGen.webSalesSchema
- case "web_returns" => TpcdsDataGen.webReturnsSchema
- case "call_center" => TpcdsDataGen.callCenterSchema
- case "catalog_page" => TpcdsDataGen.catalogPageSchema
- case "customer" => TpcdsDataGen.customerSchema
- case "customer_address" => TpcdsDataGen.customerAddressSchema
- case "customer_demographics" => TpcdsDataGen.customerDemographicsSchema
- case "date_dim" => TpcdsDataGen.dateDimSchema
- case "household_demographics" => TpcdsDataGen.householdDemographicsSchema
- case "income_band" => TpcdsDataGen.incomeBandSchema
- case "item" => TpcdsDataGen.itemSchema
- case "promotion" => TpcdsDataGen.promotionSchema
- case "reason" => TpcdsDataGen.reasonSchema
- case "ship_mode" => TpcdsDataGen.shipModeSchema
- case "store" => TpcdsDataGen.storeSchema
- case "time_dim" => TpcdsDataGen.timeDimSchema
- case "warehouse" => TpcdsDataGen.warehouseSchema
- case "web_page" => TpcdsDataGen.webPageSchema
- case "web_site" => TpcdsDataGen.webSiteSchema
- }
- val partitionBy: List[String] = if (!genPartitionedData) {
- List[String]()
- } else {
- name match {
- case "store_sales" => List("ss_sold_date_sk")
- case "catalog_sales" => List("cs_sold_date_sk")
- case "web_sales" => List("ws_sold_date_sk")
- case _ => List[String]()
- }
- }
-
- writeParquetTable(spark, name, t, schema, partitionBy)
+import org.apache.gluten.integration.table.TableLayout
+
+import org.apache.spark.sql.types.{DateType, DecimalType, LongType, StringType, StructField, StructType}
+
+case class TpcdsTableLayout(partitioned: Boolean) extends TableLayout {
+ import TpcdsTableLayout._
+
+ private val tableSchemas = Map[String, StructType](
+ "catalog_sales" -> catalogSalesSchema,
+ "catalog_returns" -> catalogReturnsSchema,
+ "inventory" -> inventorySchema,
+ "store_sales" -> storeSalesSchema,
+ "store_returns" -> storeReturnsSchema,
+ "web_sales" -> webSalesSchema,
+ "web_returns" -> webReturnsSchema,
+ "call_center" -> callCenterSchema,
+ "catalog_page" -> catalogPageSchema,
+ "customer" -> customerSchema,
+ "customer_address" -> customerAddressSchema,
+ "customer_demographics" -> customerDemographicsSchema,
+ "date_dim" -> dateDimSchema,
+ "household_demographics" -> householdDemographicsSchema,
+ "income_band" -> incomeBandSchema,
+ "item" -> itemSchema,
+ "promotion" -> promotionSchema,
+ "reason" -> reasonSchema,
+ "ship_mode" -> shipModeSchema,
+ "store" -> storeSchema,
+ "time_dim" -> timeDimSchema,
+ "warehouse" -> warehouseSchema,
+ "web_page" -> webPageSchema,
+ "web_site" -> webSiteSchema
+ )
+
+ private val partitionColumns = Map(
+ "catalog_sales" -> Seq("cs_sold_date_sk"),
+ "catalog_returns" -> Seq("cr_returned_date_sk"),
+ "store_sales" -> Seq("ss_sold_date_sk"),
+ "store_returns" -> Seq("sr_returned_date_sk"),
+ "web_sales" -> Seq("ws_sold_date_sk"),
+ "web_returns" -> Seq("wr_returned_date_sk")
+ )
+
+ override def listTableNames(): Seq[String] = {
+ tableSchemas.keys.toSeq
}
- private def writeParquetTable(
- spark: SparkSession,
- tableName: String,
- t: Table,
- schema: StructType,
- partitionBy: List[String]): Unit = {
- println(s"Generating table $tableName...")
- val rowModifier = DataGen.getRowModifier(schema, typeModifiers)
- val modifiedSchema = DataGen.modifySchema(schema, rowModifier)
-
- val stringSchema = StructType(modifiedSchema.fields.map(f => StructField(f.name, StringType)))
-
- val columns = modifiedSchema.fields.map(f => new Column(f.name).cast(f.dataType).as(f.name))
- spark
- .range(0, partitions, 1L, partitions)
- .mapPartitions {
- itr =>
- val id = itr.toArray
- if (id.length != 1) {
- throw new IllegalStateException()
- }
- val options = new Options()
- options.scale = scale
- options.parallelism = partitions
- val session = options.toSession
- val chunkSession = session.withChunkNumber(id(0).toInt + 1)
- val results = Results.constructResults(t, chunkSession).asScala.toIterator
- results.map {
- parentAndChildRow =>
- // Skip child table when generating parent table,
- // we generate every table individually no matter it is parent or child.
- val array: Array[String] = parentAndChildRow.get(0).asScala.toArray
- Row(array: _*)
- }
- }(Shim.getExpressionEncoder(stringSchema))
- .select(columns: _*)
- .write
- .format(source)
- .partitionBy(partitionBy.toArray: _*)
- .mode(SaveMode.Overwrite)
- .option("path", dir + File.separator + tableName) // storage location
- .saveAsTable(tableName)
+ override def getSchema(tableName: String): StructType = {
+ tableSchemas(tableName)
}
- override def gen(spark: SparkSession): Unit = {
- Table.getBaseTables.forEach(t => writeParquetTable(spark, t))
-
- features.foreach(feature => DataGen.Feature.run(spark, source, feature))
+ override def getPartitionColumns(tableName: String): Seq[String] = {
+ if (partitioned) {
+ partitionColumns.getOrElse(tableName, Seq.empty)
+ } else {
+ Seq.empty
+ }
}
}
-object TpcdsDataGen {
+object TpcdsTableLayout {
// generated by script
private def catalogSalesSchema = {
diff --git a/tools/gluten-it/common/src/main/scala/org/apache/gluten/integration/h/TpchSuite.scala b/tools/gluten-it/common/src/main/scala/org/apache/gluten/integration/h/TpchSuite.scala
index 142eb4e4a4..059aff4ab7 100644
--- a/tools/gluten-it/common/src/main/scala/org/apache/gluten/integration/h/TpchSuite.scala
+++ b/tools/gluten-it/common/src/main/scala/org/apache/gluten/integration/h/TpchSuite.scala
@@ -16,11 +16,12 @@
*/
package org.apache.gluten.integration.h
-import org.apache.gluten.integration.{DataGen, QuerySet, Suite, TableAnalyzer, TableCreator}
+import org.apache.gluten.integration.{DataGen, QuerySet, Suite}
import org.apache.gluten.integration.action.Action
import org.apache.gluten.integration.ds.TpcdsSuite.TPCDS_WRITE_RELATIVE_PATH
import org.apache.gluten.integration.metrics.MetricMapper
import org.apache.gluten.integration.report.TestReporter
+import org.apache.gluten.integration.table.{TableAnalyzer, TableCreator}
import org.apache.spark.SparkConf
diff --git a/tools/gluten-it/common/src/main/scala/org/apache/gluten/integration/metrics/PlanMetric.scala b/tools/gluten-it/common/src/main/scala/org/apache/gluten/integration/metrics/PlanMetric.scala
index 239bf08afe..f8f1d7761c 100644
--- a/tools/gluten-it/common/src/main/scala/org/apache/gluten/integration/metrics/PlanMetric.scala
+++ b/tools/gluten-it/common/src/main/scala/org/apache/gluten/integration/metrics/PlanMetric.scala
@@ -31,7 +31,8 @@ import scala.reflect.ClassTag
case class PlanMetric(
queryPath: String,
- plan: SparkPlan,
+ planId: Int,
+ planNodeName: String,
key: String,
metric: SQLMetric,
tags: Set[MetricTag]) {
@@ -96,8 +97,8 @@ object PlanMetric {
tr.appendRow(
Seq(
queryPath,
- m.plan.id.toString,
- m.plan.nodeName,
+ m.planId.toString,
+ m.planNodeName,
s"[${m.metric.name.getOrElse("")}] ${toNanoTime(m.metric).toString}"))
}
val out = new ByteArrayOutputStream()
@@ -119,16 +120,17 @@ object PlanMetric {
val selfTimes = metrics
.filter(_.containsTags(MetricTag.IsSelfTime))
val rows: Seq[TableRow] = selfTimes
- .groupBy(m => m.plan.id)
+ .groupBy(m => m.planId)
.toSeq
.map {
perPlanId =>
- assert(perPlanId._2.map(_.plan.id).distinct.count(_ => true) == 1)
+ assert(perPlanId._2.map(_.planId).distinct.count(_ => true) == 1)
assert(perPlanId._2.map(_.queryPath).distinct.count(_ => true) == 1)
val head = perPlanId._2.head
TableRow(
head.queryPath,
- head.plan,
+ head.planId,
+ head.planNodeName,
perPlanId._2.map(m => toNanoTime(m.metric)).sum,
perPlanId._2.map(m => (m.key, m.metric)))
}
@@ -145,7 +147,7 @@ object PlanMetric {
for (i <- 0 until (topN.min(sorted.size))) {
val row = sorted(i)
val f = new File(row.queryPath).toPath.getFileName.toString
- tr.appendRow(Seq(f, row.plan.id.toString, row.plan.nodeName, s"${row.selfTimeNs}"))
+ tr.appendRow(Seq(f, row.planId.toString, row.planNodeName, s"${row.selfTimeNs}"))
}
val out = new ByteArrayOutputStream()
tr.print(out)
@@ -157,7 +159,8 @@ object PlanMetric {
private object NodeTimeReporter {
private case class TableRow(
queryPath: String,
- plan: SparkPlan,
+ planId: Int,
+ planNodeName: String,
selfTimeNs: Long,
metrics: Seq[(String, SQLMetric)])
}
@@ -184,12 +187,12 @@ object PlanMetric {
Leaf("Selectivity"))
val probeInputNumRows = metrics
.filter(_.containsTags(MetricTag.IsJoinProbeInputNumRows))
- .groupBy(m => m.plan.id)
+ .groupBy(m => m.planId)
.toSeq
.sortBy(_._1)
val probeOutputNumRows = metrics
.filter(_.containsTags(MetricTag.IsJoinProbeOutputNumRows))
- .groupBy(m => m.plan.id)
+ .groupBy(m => m.planId)
.toSeq
.sortBy(_._1)
assert(probeInputNumRows.size == probeOutputNumRows.size)
@@ -205,7 +208,7 @@ object PlanMetric {
TableRow(
queryPath,
id1,
- inputMetrics.head.plan.nodeName,
+ inputMetrics.head.planNodeName,
inputNumRows,
outputNumRows,
selectivity)
diff --git a/tools/gluten-it/common/src/main/scala/org/apache/gluten/integration/table/AutoTableCreator.scala b/tools/gluten-it/common/src/main/scala/org/apache/gluten/integration/table/AutoTableCreator.scala
new file mode 100644
index 0000000000..4de184a0f9
--- /dev/null
+++ b/tools/gluten-it/common/src/main/scala/org/apache/gluten/integration/table/AutoTableCreator.scala
@@ -0,0 +1,83 @@
+/*
+ * 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.integration.table
+
+import org.apache.spark.sql.{AnalysisException, SparkSession}
+
+import org.apache.hadoop.fs.{FileSystem, Path}
+
+import java.net.URI
+
+import scala.collection.mutable
+import scala.collection.mutable.ArrayBuffer
+
+/** Discover tables automatically from a given file system path. */
+object AutoTableCreator extends TableCreator {
+ override def create(spark: SparkSession, source: String, dataPath: String): Unit = {
+ val uri = URI.create(dataPath)
+ val fs = FileSystem.get(uri, spark.sessionState.newHadoopConf())
+
+ val basePath = new Path(dataPath)
+ val statuses = fs.listStatus(basePath)
+
+ val tableDirs = statuses.filter(_.isDirectory).map(_.getPath)
+
+ val tableNames = ArrayBuffer[String]()
+
+ val existedTableNames = mutable.ArrayBuffer[String]()
+ val createdTableNames = mutable.ArrayBuffer[String]()
+ val recoveredPartitionTableNames = mutable.ArrayBuffer[String]()
+
+ tableDirs.foreach {
+ tablePath =>
+ val tableName = tablePath.getName
+ tableNames += tableName
+ }
+
+ println("Creating catalog tables: " + tableNames.mkString(", ") + "...")
+
+ tableDirs.foreach {
+ tablePath =>
+ val tableName = tablePath.getName
+ if (spark.catalog.tableExists(tableName)) {
+ existedTableNames += tableName
+ } else {
+ spark.catalog.createTable(tableName, tablePath.toString, source)
+ createdTableNames += tableName
+ try {
+ spark.catalog.recoverPartitions(tableName)
+ recoveredPartitionTableNames += tableName
+ } catch {
+ case _: AnalysisException =>
+ }
+ }
+ }
+
+ if (tableNames.isEmpty) {
+ return
+ }
+ if (existedTableNames.nonEmpty) {
+ println("Tables already exists: " + existedTableNames.mkString(", ") + ".")
+ }
+ if (createdTableNames.nonEmpty) {
+ println("Tables created: " + createdTableNames.mkString(", ") + ".")
+ }
+ if (recoveredPartitionTableNames.nonEmpty) {
+ println("Recovered partition tables: " + recoveredPartitionTableNames.mkString(", ") + ".")
+ }
+ }
+}
diff --git a/tools/gluten-it/common/src/main/scala/org/apache/gluten/integration/table/LayoutTableCreator.scala b/tools/gluten-it/common/src/main/scala/org/apache/gluten/integration/table/LayoutTableCreator.scala
new file mode 100644
index 0000000000..f153c0222b
--- /dev/null
+++ b/tools/gluten-it/common/src/main/scala/org/apache/gluten/integration/table/LayoutTableCreator.scala
@@ -0,0 +1,99 @@
+/*
+ * 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.integration.table
+
+import org.apache.spark.sql.{AnalysisException, Row, SparkSession}
+
+import org.apache.hadoop.fs.{FileSystem, Path}
+
+import java.net.URI
+
+import scala.collection.mutable.ArrayBuffer
+class LayoutTableCreator(layout: TableLayout) extends TableCreator {
+ override def create(spark: SparkSession, source: String, dataPath: String): Unit = {
+ val uri = URI.create(dataPath)
+ val fs = FileSystem.get(uri, spark.sessionState.newHadoopConf())
+
+ val basePath = new Path(dataPath)
+ val statuses = fs.listStatus(basePath)
+ val tableDirs = statuses.filter(_.isDirectory).map(_.getPath)
+
+ val tableNames = ArrayBuffer[String]()
+ val existedTableNames = ArrayBuffer[String]()
+ val createdTableNames = ArrayBuffer[String]()
+ val recoveredPartitionTableNames = ArrayBuffer[String]()
+
+ tableDirs.foreach(tablePath => tableNames += tablePath.getName)
+
+ println("Creating catalog tables: " + tableNames.mkString(", ") + "...")
+
+ tableDirs.foreach {
+ tablePath =>
+ val tableName = tablePath.getName
+
+ if (spark.catalog.tableExists(tableName)) {
+ existedTableNames += tableName
+ } else {
+ val schema = layout.getSchema(tableName)
+ val partitionColumns = layout.getPartitionColumns(tableName)
+
+ val nonPartitionColumns = schema.fields
+ .filterNot(f => partitionColumns.contains(f.name))
+ .map(f => s"${f.name} ${f.dataType.sql}")
+ .mkString(", ")
+
+ val partitionDDL =
+ if (partitionColumns.nonEmpty)
+ s"PARTITIONED BY (${partitionColumns.map(c => s"$c ${schema(c).dataType.sql}").mkString(", ")})"
+ else
+ ""
+
+ spark.sql(
+ s"CREATE TABLE $tableName ($nonPartitionColumns) USING $source $partitionDDL LOCATION '${tablePath.toString}'")
+
+ createdTableNames += tableName
+
+ if (partitionColumns.nonEmpty) {
+ try {
+ spark.catalog.recoverPartitions(tableName)
+ recoveredPartitionTableNames += tableName
+ } catch {
+ case ae: AnalysisException =>
+ Console.err.println(
+ s"Failed to recover partitions for table $tableName: ${ae.message}")
+ }
+ }
+ }
+ }
+
+ if (tableNames.isEmpty) {
+ return
+ }
+
+ if (existedTableNames.nonEmpty) {
+ println("Tables already exists: " + existedTableNames.mkString(", ") + ".")
+ }
+
+ if (createdTableNames.nonEmpty) {
+ println("Tables created: " + createdTableNames.mkString(", ") + ".")
+ }
+
+ if (recoveredPartitionTableNames.nonEmpty) {
+ println("Recovered partition tables: " + recoveredPartitionTableNames.mkString(", ") + ".")
+ }
+ }
+}
diff --git a/tools/gluten-it/common/src/main/scala/org/apache/gluten/integration/TableAnalyzer.scala b/tools/gluten-it/common/src/main/scala/org/apache/gluten/integration/table/TableAnalyzer.scala
similarity index 97%
rename from tools/gluten-it/common/src/main/scala/org/apache/gluten/integration/TableAnalyzer.scala
rename to tools/gluten-it/common/src/main/scala/org/apache/gluten/integration/table/TableAnalyzer.scala
index 16707aa138..1de1c7f20b 100644
--- a/tools/gluten-it/common/src/main/scala/org/apache/gluten/integration/TableAnalyzer.scala
+++ b/tools/gluten-it/common/src/main/scala/org/apache/gluten/integration/table/TableAnalyzer.scala
@@ -14,7 +14,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.gluten.integration
+package org.apache.gluten.integration.table
import org.apache.spark.sql.SparkSession
diff --git a/tools/gluten-it/common/src/main/scala/org/apache/gluten/integration/table/TableCreator.scala b/tools/gluten-it/common/src/main/scala/org/apache/gluten/integration/table/TableCreator.scala
new file mode 100644
index 0000000000..1af92ae368
--- /dev/null
+++ b/tools/gluten-it/common/src/main/scala/org/apache/gluten/integration/table/TableCreator.scala
@@ -0,0 +1,33 @@
+/*
+ * 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.integration.table
+
+import org.apache.spark.sql.SparkSession
+
+trait TableCreator {
+ def create(spark: SparkSession, source: String, dataPath: String): Unit
+}
+
+object TableCreator {
+ def discoverFromFiles(): TableCreator = {
+ AutoTableCreator
+ }
+
+ def createFromLayout(layout: TableLayout): TableCreator = {
+ new LayoutTableCreator(layout)
+ }
+}
diff --git a/tools/gluten-it/common/src/main/scala/org/apache/gluten/integration/table/TableLayout.scala b/tools/gluten-it/common/src/main/scala/org/apache/gluten/integration/table/TableLayout.scala
new file mode 100644
index 0000000000..66f02237ed
--- /dev/null
+++ b/tools/gluten-it/common/src/main/scala/org/apache/gluten/integration/table/TableLayout.scala
@@ -0,0 +1,25 @@
+/*
+ * 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.integration.table
+
+import org.apache.spark.sql.types.StructType
+
+trait TableLayout {
+ def listTableNames(): Seq[String]
+ def getSchema(tableName: String): StructType
+ def getPartitionColumns(tableName: String): Seq[String]
+}
diff --git a/tools/gluten-it/common/src/main/scala/org/apache/spark/sql/SparkQueryRunner.scala b/tools/gluten-it/common/src/main/scala/org/apache/spark/sql/SparkQueryRunner.scala
index 6a5117be83..dcd8fcbb6d 100644
--- a/tools/gluten-it/common/src/main/scala/org/apache/spark/sql/SparkQueryRunner.scala
+++ b/tools/gluten-it/common/src/main/scala/org/apache/spark/sql/SparkQueryRunner.scala
@@ -17,24 +17,24 @@
package org.apache.spark.sql
import org.apache.gluten.integration.Query
-import org.apache.gluten.integration.metrics.{MetricMapper, MetricTag, PlanMetric}
+import org.apache.gluten.integration.metrics.{MetricMapper, PlanMetric}
import org.apache.spark.{SparkContext, Success, TaskKilled}
import org.apache.spark.executor.ExecutorMetrics
-import org.apache.spark.scheduler.{SparkListener, SparkListenerExecutorMetricsUpdate, SparkListenerTaskEnd, SparkListenerTaskStart}
-import org.apache.spark.sql.KillTaskListener.INIT_WAIT_TIME_MS
+import org.apache.spark.scheduler.{SparkListener, SparkListenerEvent, SparkListenerExecutorMetricsUpdate, SparkListenerTaskEnd, SparkListenerTaskStart}
import org.apache.spark.sql.catalyst.QueryPlanningTracker
import org.apache.spark.sql.execution.{QueryExecution, SparkPlan}
import org.apache.spark.sql.execution.adaptive.{AdaptiveSparkPlanExec, QueryStageExec}
import org.apache.spark.sql.execution.exchange.ReusedExchangeExec
+import org.apache.spark.sql.execution.ui.{SparkListenerSQLExecutionEnd, SparkListenerSQLExecutionStart}
-import com.google.common.base.Preconditions
import org.apache.commons.lang3.RandomUtils
-import java.io.ByteArrayOutputStream
-import java.nio.charset.StandardCharsets
+import java.util.concurrent.ConcurrentHashMap
import java.util.concurrent.atomic.AtomicInteger
+import scala.collection.JavaConverters._
+import scala.collection.concurrent.TrieMap
import scala.collection.mutable
object SparkQueryRunner {
@@ -74,11 +74,19 @@ object SparkQueryRunner {
val sc = spark.sparkContext
sc.setJobDescription(desc)
+ // Wait until all SparkListener events are processed before we start executing the query, to make sure
+ // the metrics we collect are accurate.
+ sc.listenerBus.waitUntilEmpty()
+
// Executor metrics listener.
val em = new ExecutorMetrics()
val metricsListener = new MetricsListener(em)
sc.addSparkListener(metricsListener)
+ // Execution time listener.
+ val executionTimeListener = new ExecutionTimeListener()
+ sc.addSparkListener(executionTimeListener)
+
// kill task listener.
val killTaskListener: Option[KillTaskListener] = if (randomKillTasks) {
Some(new KillTaskListener(sc))
@@ -89,33 +97,28 @@ object SparkQueryRunner {
println(s"Executing SQL query from resource path ${query.path}...")
try {
- val tracker = new QueryPlanningTracker
- val prev = System.nanoTime()
+ val otherTracker = new QueryPlanningTracker
val df = spark.sql(query.sql)
- val rows = QueryPlanningTracker.withTracker(tracker) {
+ val rows = QueryPlanningTracker.withTracker(otherTracker) {
df.collect()
}
- val totalMillis = (System.nanoTime() - prev) / 1000000L
if (explain) {
df.explain(extended = true)
}
+ // TODO: sparkTracker and otherTracker include planning time information,
+ // while they are not accurate enough so we paused on using them for now.
val sparkTracker = df.queryExecution.tracker
- val sparkRulesMillis =
- sparkTracker.rules.map(_._2.totalTimeNs).sum / 1000000L
- val otherRulesMillis =
- tracker.rules.map(_._2.totalTimeNs).sum / 1000000L
- val planMillis = sparkRulesMillis + otherRulesMillis
+ // This makes sure all the SparkListenerSQLExecutionEnd event is processed before we
+ // get the execution time and metrics.
+ sc.listenerBus.waitUntilEmpty()
+ val totalMillis = executionTimeListener.wallMillis()
val collectedExecutorMetrics =
executorMetrics.map(name => (name, em.getMetricValue(name))).toMap
val collectedSQLMetrics = collectSQLMetrics(query.path, metricMapper, df.queryExecution)
- RunResult(
- rows,
- planMillis,
- totalMillis - planMillis,
- collectedSQLMetrics,
- collectedExecutorMetrics)
+ RunResult(rows, totalMillis, collectedSQLMetrics, collectedExecutorMetrics)
} finally {
sc.removeSparkListener(metricsListener)
+ sc.removeSparkListener(executionTimeListener)
killTaskListener.foreach(
l => {
sc.removeSparkListener(l)
@@ -153,7 +156,7 @@ object SparkQueryRunner {
p.metrics.map {
case keyValue @ (k, m) =>
val tags = mapper.map(p, k, m)
- PlanMetric(queryPath, p, k, m, tags.toSet)
+ PlanMetric(queryPath, p.id, p.nodeName, k, m, tags.toSet)
}
}
all.toSeq
@@ -162,7 +165,6 @@ object SparkQueryRunner {
case class RunResult(
rows: Seq[Row],
- planningTimeMillis: Long,
executionTimeMillis: Long,
sqlMetrics: Seq[PlanMetric],
executorMetrics: Map[String, Long])
@@ -201,7 +203,7 @@ class KillTaskListener(val sc: SparkContext) extends SparkListener {
val total = Math.min(
stageKillMaxWaitTimeLookup.computeIfAbsent(taskStart.stageId, _ => Long.MaxValue),
stageKillWaitTimeLookup
- .computeIfAbsent(taskStart.stageId, _ => INIT_WAIT_TIME_MS)
+ .computeIfAbsent(taskStart.stageId, _ => KillTaskListener.INIT_WAIT_TIME_MS)
)
val elapsed = System.currentTimeMillis() - startMs
val remaining = total - elapsed
@@ -263,3 +265,24 @@ class KillTaskListener(val sc: SparkContext) extends SparkListener {
object KillTaskListener {
private val INIT_WAIT_TIME_MS: Long = 50L
}
+
+class ExecutionTimeListener extends SparkListener {
+ private val executionStart = new ConcurrentHashMap[Long, Long]().asScala
+ private val executionEnd = new ConcurrentHashMap[Long, Long]().asScala
+
+ override def onOtherEvent(event: SparkListenerEvent): Unit = event match {
+ case e: SparkListenerSQLExecutionStart =>
+ executionStart(e.executionId) = e.time
+ case e: SparkListenerSQLExecutionEnd =>
+ executionEnd(e.executionId) = e.time
+ case _ =>
+ }
+
+ def wallMillis(): Long = {
+ executionEnd.map {
+ case (executionId, endTime) =>
+ val startTime = executionStart(executionId)
+ endTime - startTime
+ }.sum
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org