diff --git a/build.sbt b/build.sbt
index 2303e62..f7dbef6 100644
--- a/build.sbt
+++ b/build.sbt
@@ -5,7 +5,9 @@ name := "spark-sql-perf"
 
 organization := "com.databricks"
 
-scalaVersion := "2.12.10"
+// for scala 2.12 use  2.12.17 version
+// for scala 2.13 use 2.13.10
+scalaVersion := "2.13.16"
 
 crossScalaVersions := Seq("2.12.10")
 
@@ -14,11 +16,12 @@ sparkPackageName := "databricks/spark-sql-perf"
 // All Spark Packages need a license
 licenses := Seq("Apache-2.0" -> url("http://opensource.org/licenses/Apache-2.0"))
 
-sparkVersion := "3.0.0"
+// for scala 2.12 use spark 3.4.0
+// for scala 2.13 use spark 4.0.0-SNAPSHOT
+sparkVersion := "4.1.1"
 
 sparkComponents ++= Seq("sql", "hive", "mllib")
-
-
+resolvers += Resolver.mavenLocal
 initialCommands in console :=
   """
     |import org.apache.spark.sql._
@@ -34,9 +37,9 @@ initialCommands in console :=
 
 libraryDependencies += "com.github.scopt" %% "scopt" % "3.7.1"
 
-libraryDependencies += "com.twitter" %% "util-jvm" % "6.45.0" % "provided"
+libraryDependencies += "com.twitter" %% "util-jvm" % "22.7.0" % "provided"
 
-libraryDependencies += "org.scalatest" %% "scalatest" % "3.0.5" % "test"
+libraryDependencies += "org.scalatest" %% "scalatest" % "3.2.16" % "test"
 
 libraryDependencies += "org.yaml" % "snakeyaml" % "1.23"
 
diff --git a/src/main/scala/com/databricks/spark/sql/perf/Benchmark.scala b/src/main/scala/com/databricks/spark/sql/perf/Benchmark.scala
index ebb4935..ba82ed0 100644
--- a/src/main/scala/com/databricks/spark/sql/perf/Benchmark.scala
+++ b/src/main/scala/com/databricks/spark/sql/perf/Benchmark.scala
@@ -23,14 +23,14 @@ import scala.concurrent.duration._
 import scala.language.implicitConversions
 import scala.util.{Success, Try, Failure => SFailure}
 import scala.util.control.NonFatal
-
 import org.apache.spark.rdd.RDD
-import org.apache.spark.sql.{Dataset, DataFrame, SQLContext, SparkSession}
+import org.apache.spark.sql.{DataFrame, Dataset, SQLContext, SparkSession}
 import org.apache.spark.sql.catalyst.analysis.UnresolvedRelation
 import org.apache.spark.SparkContext
-
 import com.databricks.spark.sql.perf.cpu._
 
+import java.io.{PrintWriter, StringWriter}
+
 /**
  * A collection of queries that test a particular aspect of Spark SQL.
  *
@@ -212,7 +212,7 @@ abstract class Benchmark(
       new SparkPerfExecution(
         name,
         Map.empty,
-        () => Unit,
+        () => (),
         () => rdd.count(),
         rdd.toDebugString)
     }
@@ -250,11 +250,14 @@ abstract class Benchmark(
           executionTime = Some(timeMs))
       } catch {
         case e: Exception =>
+          val sw: StringWriter = new StringWriter();
+          val pw: PrintWriter  = new PrintWriter(sw);
+          e.printStackTrace(pw);
           BenchmarkResult(
             name = name,
             mode = executionMode.toString,
             parameters = parameters,
-            failure = Some(Failure(e.getClass.getSimpleName, e.getMessage)))
+            failure = Some(Failure(e.getClass.getSimpleName, e.getMessage + "..." + sw.toString)))
       }
     }
   }
@@ -428,8 +431,11 @@ object Benchmark {
       }
 
       try {
-        val resultsTable = sqlContext.createDataFrame(results)
-        logMessage(s"Results written to table: 'sqlPerformance' at $resultPath")
+        import scala.jdk.CollectionConverters._
+        import sqlContext.implicits._
+        val resultsTable = sqlContext.createDataset(results.asJava)
+        logMessage(s"Results written to table: 'sqlPerformance' at $resultPath with schema" +
+          s" ${resultsTable.schema.toString()}")
         resultsTable
           .coalesce(1)
           .write
@@ -475,14 +481,18 @@ object Benchmark {
 
     /** Returns results from an actively running experiment. */
     def getCurrentResults() = {
-      val tbl = sqlContext.createDataFrame(currentResults)
+      import scala.jdk.CollectionConverters._
+      import sqlContext.implicits._
+      val tbl = sqlContext.createDataset(currentResults.asJava)
       tbl.createOrReplaceTempView("currentResults")
       tbl
     }
 
     /** Returns full iterations from an actively running experiment. */
     def getCurrentRuns() = {
-      val tbl = sqlContext.createDataFrame(currentRuns)
+      import scala.jdk.CollectionConverters._
+      import sqlContext.implicits._
+      val tbl = sqlContext.createDataset(currentRuns.asJava)
       tbl.createOrReplaceTempView("currentRuns")
       tbl
     }
diff --git a/src/main/scala/com/databricks/spark/sql/perf/Benchmarkable.scala b/src/main/scala/com/databricks/spark/sql/perf/Benchmarkable.scala
index 24efef7..7e0b428 100644
--- a/src/main/scala/com/databricks/spark/sql/perf/Benchmarkable.scala
+++ b/src/main/scala/com/databricks/spark/sql/perf/Benchmarkable.scala
@@ -85,7 +85,7 @@ trait Benchmarkable {
               mode = executionMode.toString,
               parameters = Map.empty,
               failure = Some(Failure(e.getClass.getSimpleName,
-                e.getMessage + ":\n" + e.getStackTraceString)))
+                e.getMessage + ":\n" + e.getStackTrace.toString)))
         }
       }
     }
diff --git a/src/main/scala/com/databricks/spark/sql/perf/DatasetPerformance.scala b/src/main/scala/com/databricks/spark/sql/perf/DatasetPerformance.scala
index 0aaa629..b3d25d4 100644
--- a/src/main/scala/com/databricks/spark/sql/perf/DatasetPerformance.scala
+++ b/src/main/scala/com/databricks/spark/sql/perf/DatasetPerformance.scala
@@ -133,7 +133,7 @@ class DatasetPerformance extends Benchmark {
     new SparkPerfExecution(
       "RDD: average",
       Map.empty,
-      prepare = () => Unit,
+      prepare = () => (),
       run = () => {
         val sumAndCount =
           smallrdd.map(i => (i, 1)).reduce((a, b) => (a._1 + b._1, a._2 + b._2))
diff --git a/src/main/scala/com/databricks/spark/sql/perf/Query.scala b/src/main/scala/com/databricks/spark/sql/perf/Query.scala
index babc63f..0757af1 100644
--- a/src/main/scala/com/databricks/spark/sql/perf/Query.scala
+++ b/src/main/scala/com/databricks/spark/sql/perf/Query.scala
@@ -19,11 +19,12 @@ package com.databricks.spark.sql.perf
 import scala.collection.mutable
 import scala.collection.mutable.ArrayBuffer
 import scala.language.implicitConversions
-
 import org.apache.spark.sql.DataFrame
 import org.apache.spark.sql.catalyst.analysis.UnresolvedRelation
 import org.apache.spark.sql.execution.SparkPlan
 
+import java.io.{PrintWriter, StringWriter}
+
 
 /** Holds one benchmark query and its metadata. */
 class Query(
@@ -92,7 +93,7 @@ class Query(
             messages += s"Breakdown: ${node.simpleString(maxFields)}"
             val newNode = buildDataFrame.queryExecution.executedPlan.p(index)
             val executionTime = measureTimeMs {
-              newNode.execute().foreach((row: Any) => Unit)
+              newNode.execute().foreach((row: Any) => ())
             }
             timeMap += ((index, executionTime))
 
@@ -117,14 +118,22 @@ class Query(
       // Note: queryExecution.{logical, analyzed, optimizedPlan, executedPlan} has been already
       // lazily evaluated above, so below we will count only execution time.
       var result: Option[Long] = None
+      var numRowsOutput = 0
       val executionTime = measureTimeMs {
         executionMode match {
-          case ExecutionMode.CollectResults => dataFrame.collect()
-          case ExecutionMode.ForeachResults => dataFrame.foreach { _ => ():Unit }
+          case ExecutionMode.CollectResults =>
+               val x = dataFrame.collect()
+               numRowsOutput = x.length
+
+          case ExecutionMode.ForeachResults =>
+            numRowsOutput = dataFrame.collect().length
+            dataFrame.foreach { _ => ():Unit }
+
           case ExecutionMode.WriteParquet(location) =>
             dataFrame.write.parquet(s"$location/$name.parquet")
           case ExecutionMode.HashResults =>
             // SELECT SUM(CRC32(CONCAT_WS(", ", *))) FROM (benchmark query)
+            numRowsOutput = dataFrame.collect().length
             val row =
               dataFrame
                 .selectExpr(s"sum(crc32(concat_ws(',', *)))")
@@ -149,13 +158,17 @@ class Query(
         executionTime = executionTime,
         result = result,
         queryExecution = dataFrame.queryExecution.toString,
-        breakDown = breakdownResults)
+        breakDown = breakdownResults,
+        numRows = numRowsOutput)
     } catch {
       case e: Exception =>
+        val sw: StringWriter = new StringWriter();
+        val pw: PrintWriter = new PrintWriter(sw);
+        e.printStackTrace(pw);
          BenchmarkResult(
            name = name,
            mode = executionMode.toString,
-           failure = Failure(e.getClass.getName, e.getMessage))
+           failure = Failure(e.getClass.getName, e.getMessage + "...." + sw.toString))
     }
   }
 
diff --git a/src/main/scala/com/databricks/spark/sql/perf/Tables.scala b/src/main/scala/com/databricks/spark/sql/perf/Tables.scala
index 177d38c..a36fbee 100644
--- a/src/main/scala/com/databricks/spark/sql/perf/Tables.scala
+++ b/src/main/scala/com/databricks/spark/sql/perf/Tables.scala
@@ -174,10 +174,40 @@ abstract class Tables(sqlContext: SQLContext, scaleFactor: String,
       overwrite: Boolean,
       clusterByPartitionColumns: Boolean,
       filterOutNullPartitionValues: Boolean,
-      numPartitions: Int): Unit = {
+      numPartitions: Int,
+      sortOnCol: Boolean): Unit = {
       val mode = if (overwrite) SaveMode.Overwrite else SaveMode.Ignore
 
-      val data = df(format != "text", numPartitions)
+      val dataTemp = df(format != "text", numPartitions)
+      val data = if (sortOnCol) {
+        if (name.toLowerCase.contains("store_sales")) {
+          dataTemp.sortWithinPartitions("ss_sold_date_sk")
+        } else if (name.toLowerCase.contains("web_sales")) {
+          dataTemp.sortWithinPartitions("ws_sold_date_sk")
+        } else if (name.toLowerCase.contains("catalog_sales")) {
+          dataTemp.sortWithinPartitions("cs_sold_date_sk")
+        } else if (name.toLowerCase.contains("catalog_returns")) {
+          dataTemp.sortWithinPartitions("cr_returned_date_sk")
+        } else if (name.toLowerCase.contains("inventory")) {
+          dataTemp.sortWithinPartitions("inv_date_sk")
+        }
+
+        else if (name.toLowerCase.contains("store_returns")) {
+          dataTemp.sortWithinPartitions("sr_returned_date_sk")
+        }
+
+        else if (name.toLowerCase.contains("web_returns")) {
+          dataTemp.sortWithinPartitions("wr_returned_date_sk")
+        }
+
+
+        else {
+          dataTemp
+        }
+      } else {
+        dataTemp
+      }
+
       val tempTableName = s"${name}_text"
       data.createOrReplaceTempView(tempTableName)
 
@@ -263,12 +293,90 @@ abstract class Tables(sqlContext: SQLContext, scaleFactor: String,
       }
     }
 
+    def  createInternalHiveTable(location: String, format: String, databaseName: String,
+        overwrite: Boolean, numSplits: Option[Int], discoverPartitions: Boolean = true,
+        sortOnCol: Boolean): Unit = {
+
+      val qualifiedTableName = databaseName + "." + name
+      val tableExists = sqlContext.tableNames(databaseName).contains(name)
+      if (overwrite) {
+        sqlContext.sql(s"DROP TABLE IF EXISTS $databaseName.$name")
+      }
+      if (!tableExists || overwrite) {
+        println(s"Creating internal table $name in database $databaseName using data stored in $location.")
+        log.info(s"Creating internal table $name in database $databaseName using data stored in $location.")
+        val temp = qualifiedTableName + "_temp"
+        val tempdf = sqlContext.createExternalTable(temp, location, format)
+        val df = numSplits.map(x => tempdf.repartition(x)).getOrElse(tempdf)
+        val sortedDf = if (sortOnCol) {
+          if (qualifiedTableName.toLowerCase.contains("store_sales")) {
+            df.sortWithinPartitions("ss_sold_date_sk") //.orderBy("ss_item_sk").orderBy("ss_store_sk")
+          } else if (qualifiedTableName.toLowerCase.contains("web_sales")) {
+            df.sortWithinPartitions("ws_sold_date_sk")
+          } else if (qualifiedTableName.toLowerCase.contains("catalog_sales")) {
+            df.sortWithinPartitions("cs_sold_date_sk")
+          } else {
+            df
+          }
+        } else {
+          df
+        }
+
+        sortedDf.write.format("parquet").mode(SaveMode.Overwrite).saveAsTable(qualifiedTableName)
+
+        sqlContext.sql(s"DROP TABLE IF EXISTS $temp")
+      }
+      if (partitionColumns.nonEmpty && discoverPartitions) {
+        throw new UnsupportedOperationException("not implemented partitiooned clause")
+      }
+    }
+
+    def  createIcebergTable(location: String, format: String, databaseName: String,
+      overwrite: Boolean, numSplits: Option[Int], discoverPartitions: Boolean = false,
+      sortOnCol: Boolean): Unit = {
+
+      val qualifiedTableName = databaseName + "." + name
+      val tableExists = sqlContext.tableNames(databaseName).contains(name)
+      if (overwrite) {
+        sqlContext.sql(s"DROP TABLE IF EXISTS $databaseName.$name")
+      }
+      if (!tableExists || overwrite) {
+        println(s"Creating internal table $name in database $databaseName using data stored in $location.")
+        log.info(s"Creating internal table $name in database $databaseName using data stored in $location.")
+        val temp = qualifiedTableName + "_temp"
+        val tempdf = sqlContext.createExternalTable(temp, location, format)
+        val df = numSplits.map(x => tempdf.repartition(x)).getOrElse(tempdf)
+        val sortedDf = if (sortOnCol) {
+          if (qualifiedTableName.toLowerCase.contains("store_sales")) {
+            df.sortWithinPartitions("ss_sold_date_sk") //.orderBy("ss_item_sk").orderBy("ss_store_sk")
+          } else if (qualifiedTableName.toLowerCase.contains("web_sales")) {
+            df.sortWithinPartitions("ws_sold_date_sk")
+          } else if (qualifiedTableName.toLowerCase.contains("catalog_sales")) {
+            df.sortWithinPartitions("cs_sold_date_sk")
+          } else {
+            df
+          }
+        } else {
+          df
+        }
+
+        sortedDf.writeTo(qualifiedTableName).tableProperty("write.format.default", "parquet")
+          .using("iceberg").create
+
+        sqlContext.sql(s"DROP TABLE IF EXISTS $temp")
+      }
+      if (partitionColumns.nonEmpty && discoverPartitions) {
+        throw new UnsupportedOperationException("not implemented partitiooned clause")
+      }
+    }
+
     def createTemporaryTable(location: String, format: String): Unit = {
       println(s"Creating temporary table $name using data stored in $location.")
       log.info(s"Creating temporary table $name using data stored in $location.")
       sqlContext.read.format(format).load(location).createOrReplaceTempView(name)
     }
 
+
     def analyzeTable(databaseName: String, analyzeColumns: Boolean = false): Unit = {
       println(s"Analyzing table $name.")
       log.info(s"Analyzing table $name.")
@@ -290,7 +398,8 @@ abstract class Tables(sqlContext: SQLContext, scaleFactor: String,
       clusterByPartitionColumns: Boolean,
       filterOutNullPartitionValues: Boolean,
       tableFilter: String = "",
-      numPartitions: Int = 100): Unit = {
+      numPartitions: Int = 100,
+      sortOnCol: Boolean = true): Unit = {
     var tablesToBeGenerated = if (partitionTables) {
       tables
     } else {
@@ -307,7 +416,7 @@ abstract class Tables(sqlContext: SQLContext, scaleFactor: String,
     tablesToBeGenerated.foreach { table =>
       val tableLocation = s"$location/${table.name}"
       table.genData(tableLocation, format, overwrite, clusterByPartitionColumns,
-        filterOutNullPartitionValues, numPartitions)
+        filterOutNullPartitionValues, numPartitions, sortOnCol)
     }
   }
 
@@ -330,6 +439,50 @@ abstract class Tables(sqlContext: SQLContext, scaleFactor: String,
     log.info(s"The current database has been set to $databaseName.")
   }
 
+  def createIcebergTables(location: String, format: String, databaseName: String,
+    overwrite: Boolean, discoverPartitions: Boolean, tableFilter: String = "",
+                          numSplits: Option[Int] = None,
+                          sortOnCol: Boolean = true): Unit = {
+
+    val filtered = if (tableFilter.isEmpty) {
+      tables
+    } else {
+      tables.filter(_.name == tableFilter)
+    }
+
+    sqlContext.sql(s"CREATE DATABASE IF NOT EXISTS $databaseName")
+    filtered.foreach { table =>
+      val tableLocation = s"$location/${table.name}"
+      table.createIcebergTable(tableLocation, format, databaseName, overwrite,
+        numSplits, discoverPartitions, sortOnCol)
+    }
+    sqlContext.sql(s"USE $databaseName")
+    println(s"The current database has been set to $databaseName.")
+    log.info(s"The current database has been set to $databaseName.")
+  }
+
+  def createInternalTables(location: String, format: String, databaseName: String,
+      overwrite: Boolean, discoverPartitions: Boolean, tableFilter: String = "",
+      numSplits: Option[Int] = None,
+      sortOnCol: Boolean = true): Unit = {
+
+    val filtered = if (tableFilter.isEmpty) {
+      tables
+    } else {
+      tables.filter(_.name == tableFilter)
+    }
+
+    sqlContext.sql(s"CREATE DATABASE IF NOT EXISTS $databaseName")
+    filtered.foreach { table =>
+      val tableLocation = s"$location/${table.name}"
+      table.createInternalHiveTable(tableLocation, format, databaseName, overwrite,
+        numSplits, discoverPartitions, sortOnCol)
+    }
+    sqlContext.sql(s"USE $databaseName")
+    println(s"The current database has been set to $databaseName.")
+    log.info(s"The current database has been set to $databaseName.")
+  }
+
   def createTemporaryTables(location: String, format: String, tableFilter: String = ""): Unit = {
     val filtered = if (tableFilter.isEmpty) {
       tables
diff --git a/src/main/scala/com/databricks/spark/sql/perf/mllib/MLLib.scala b/src/main/scala/com/databricks/spark/sql/perf/mllib/MLLib.scala
index c0bf70e..0a93e52 100644
--- a/src/main/scala/com/databricks/spark/sql/perf/mllib/MLLib.scala
+++ b/src/main/scala/com/databricks/spark/sql/perf/mllib/MLLib.scala
@@ -36,7 +36,7 @@ object MLLib {
       executionsToRun = benchmarks)
     e.waitForFinish(1000 * 60 * 30)
     logger.info("Run finished")
-    e.getCurrentResults()
+    e.getCurrentResults().toDF()
   }
 
   private def getConfig(resourcePath: String): String = {
@@ -97,6 +97,6 @@ object MLLib {
       forkThread = false)
     e.waitForFinish(conf.timeout.toSeconds.toInt)
     logger.info("Run finished")
-    e.getCurrentResults()
+    e.getCurrentResults().toDF()
   }
 }
diff --git a/src/main/scala/com/databricks/spark/sql/perf/mllib/MLPipelineStageBenchmarkable.scala b/src/main/scala/com/databricks/spark/sql/perf/mllib/MLPipelineStageBenchmarkable.scala
index 8296f46..80d7f6e 100644
--- a/src/main/scala/com/databricks/spark/sql/perf/mllib/MLPipelineStageBenchmarkable.scala
+++ b/src/main/scala/com/databricks/spark/sql/perf/mllib/MLPipelineStageBenchmarkable.scala
@@ -37,7 +37,7 @@ class MLPipelineStageBenchmarkable(
       trainingData.count()
     } catch {
       case NonFatal(e) =>
-        println(s"$this error in beforeBenchmark: ${e.getStackTraceString}")
+        println(s"$this error in beforeBenchmark: ${e.getStackTrace.toString}")
         throw e
     }
   }
@@ -103,7 +103,7 @@ class MLPipelineStageBenchmarkable(
           mode = executionMode.toString,
           parameters = params.toMap,
           failure = Some(Failure(e.getClass.getSimpleName,
-            e.getMessage + ":\n" + e.getStackTraceString)))
+            e.getMessage + ":\n" + e.getStackTrace.toString)))
     } finally {
       Option(testData).map(_.unpersist())
       Option(trainingData).map(_.unpersist())
diff --git a/src/main/scala/com/databricks/spark/sql/perf/results.scala b/src/main/scala/com/databricks/spark/sql/perf/results.scala
index 28d7226..bb7a421 100644
--- a/src/main/scala/com/databricks/spark/sql/perf/results.scala
+++ b/src/main/scala/com/databricks/spark/sql/perf/results.scala
@@ -90,7 +90,8 @@ case class BenchmarkResult(
     queryExecution: Option[String] = None,
     failure: Option[Failure] = None,
     mlResult: Option[Array[MLMetric]] = None,
-    benchmarkId: Option[String] = None)
+    benchmarkId: Option[String] = None,
+    numRows: Int = 0)
 
 /**
  * The execution time of a subtree of the query plan tree of a specific query.
diff --git a/src/main/scala/com/databricks/spark/sql/perf/tpcds/TPCDS_2_4_Queries.scala b/src/main/scala/com/databricks/spark/sql/perf/tpcds/TPCDS_2_4_Queries.scala
index f78dfe0..abc27ab 100644
--- a/src/main/scala/com/databricks/spark/sql/perf/tpcds/TPCDS_2_4_Queries.scala
+++ b/src/main/scala/com/databricks/spark/sql/perf/tpcds/TPCDS_2_4_Queries.scala
@@ -27,7 +27,8 @@ trait Tpcds_2_4_Queries extends Benchmark {
 
   import ExecutionMode._
 
-  val queryNames = Seq(
+
+  val queryNames = if (true )  {Seq(
     "q1", "q2", "q3", "q4", "q5", "q6", "q7", "q8", "q9", "q10",
     "q11", "q12", "q13", "q14a", "q14b", "q15", "q16", "q17", "q18", "q19",
     "q20", "q21", "q22", "q23a", "q23b", "q24a", "q24b", "q25", "q26", "q27",
@@ -41,6 +42,11 @@ trait Tpcds_2_4_Queries extends Benchmark {
     "ss_max"
   )
 
+}else {
+  Seq("q61") //59//58
+}
+
+
   val tpcds2_4Queries = queryNames.map { queryName =>
     val queryContent: String = IOUtils.toString(
       getClass().getClassLoader().getResourceAsStream(s"tpcds_2_4/$queryName.sql"))
diff --git a/src/main/scala/org/apache/spark/ml/ModelBuilderSSP.scala b/src/main/scala/org/apache/spark/ml/ModelBuilderSSP.scala
index fa66e00..99cef6d 100644
--- a/src/main/scala/org/apache/spark/ml/ModelBuilderSSP.scala
+++ b/src/main/scala/org/apache/spark/ml/ModelBuilderSSP.scala
@@ -81,20 +81,20 @@ object TreeBuilder {
    * @param numClasses  Number of classes.
    */
   private class ClassLabelPairGenerator(val numClasses: Int)
-    extends RandomDataGenerator[Pair[Double, Double]] {
+    extends RandomDataGenerator[(Double, Double)] {
 
     require(numClasses >= 2,
       s"ClassLabelPairGenerator given label numClasses = $numClasses, but numClasses should be >= 2.")
 
     private val rng = new java.util.Random()
 
-    override def nextValue(): Pair[Double, Double] = {
+    override def nextValue(): (Double, Double) = {
       val left = rng.nextInt(numClasses)
       var right = rng.nextInt(numClasses)
       while (right == left) {
         right = rng.nextInt(numClasses)
       }
-      new Pair[Double, Double](left, right)
+      (left, right)
     }
 
     override def setSeed(seed: Long): Unit = {
@@ -109,12 +109,12 @@ object TreeBuilder {
    * Generator for a pair of real-valued labels.
    * Pairs are useful for trees to make sure sibling leaf nodes make different predictions.
    */
-  private class RealLabelPairGenerator() extends RandomDataGenerator[Pair[Double, Double]] {
+  private class RealLabelPairGenerator() extends RandomDataGenerator[(Double, Double)] {
 
     private val rng = new java.util.Random()
 
-    override def nextValue(): Pair[Double, Double] =
-      new Pair[Double, Double](rng.nextDouble(), rng.nextDouble())
+    override def nextValue(): (Double, Double) =
+      (rng.nextDouble(), rng.nextDouble())
 
     override def setSeed(seed: Long): Unit = {
       rng.setSeed(seed)
@@ -185,7 +185,7 @@ object TreeBuilder {
       subtreeDepth: Int,
       featureArity: Array[Int],
       impurityCalculator: ImpurityCalculator,
-      labelGenerator: RandomDataGenerator[Pair[Double, Double]],
+      labelGenerator: RandomDataGenerator[(Double, Double)],
       usedFeatures: Set[Int],
       rng: scala.util.Random): Node = {