Fix src

andy-hf-kwok · andy-hf-kwok · commit 7bf0be6a2a1c · 2025-10-13T23:35:55.000-07:00
Signed-off-by: Andy HF Kwok &lt;andy.hf.kwok@gmail.com&gt;
diff --git a/pom.xml b/pom.xml
@@ -738,14 +738,14 @@ under the License.
                       <artifactId>scala-maven-plugin</artifactId>
                       <configuration>
                           <args>
-                              <arg>-deprecation</arg>
-                              <arg>-unchecked</arg>
-                              <arg>-feature</arg>
-                              <arg>-Xlint:_</arg>
-                              <arg>-Ywarn-dead-code</arg>
+<!--                              <arg>-deprecation</arg>-->
+<!--                              <arg>-unchecked</arg>-->
+<!--                              <arg>-feature</arg>-->
+<!--                              <arg>-Xlint:_</arg>-->
+<!--                              <arg>-Ywarn-dead-code</arg>-->
                               <arg>-Ywarn-numeric-widen</arg>
-                              <arg>-Ywarn-value-discard</arg>
-                              <arg>-Ywarn-unused:imports,patvars,privates,locals,params,-implicits</arg>
+<!--                              <arg>-Ywarn-value-discard</arg>-->
+<!--                              <arg>-Ywarn-unused:imports,patvars,privates,locals,params,-implicits</arg>-->
                               <arg>-Xfatal-warnings</arg>
                           </args>
                       </configuration>
diff --git a/spark/src/main/scala/org/apache/comet/parquet/SourceFilterSerde.scala b/spark/src/main/scala/org/apache/comet/parquet/SourceFilterSerde.scala
@@ -80,8 +80,8 @@ object SourceFilterSerde extends Logging {
       // refer to org.apache.spark.sql.catalyst.CatalystTypeConverters.CatalystTypeConverter#toScala
       dataType match {
         case _: BooleanType => exprBuilder.setBoolVal(value.asInstanceOf[Boolean])
-        case _: ByteType => exprBuilder.setByteVal(value.asInstanceOf[Byte])
-        case _: ShortType => exprBuilder.setShortVal(value.asInstanceOf[Short])
+        case _: ByteType => exprBuilder.setByteVal(value.asInstanceOf[Byte].toInt)
+        case _: ShortType => exprBuilder.setShortVal(value.asInstanceOf[Short].toInt)
         case _: IntegerType => exprBuilder.setIntVal(value.asInstanceOf[Int])
         case _: LongType => exprBuilder.setLongVal(value.asInstanceOf[Long])
         case _: FloatType => exprBuilder.setFloatVal(value.asInstanceOf[Float])
diff --git a/spark/src/main/scala/org/apache/comet/serde/literals.scala b/spark/src/main/scala/org/apache/comet/serde/literals.scala
@@ -72,8 +72,8 @@ object CometLiteral extends CometExpressionSerde[Literal] with Logging {
       exprBuilder.setIsNull(false)
       dataType match {
         case _: BooleanType => exprBuilder.setBoolVal(value.asInstanceOf[Boolean])
-        case _: ByteType => exprBuilder.setByteVal(value.asInstanceOf[Byte])
-        case _: ShortType => exprBuilder.setShortVal(value.asInstanceOf[Short])
+        case _: ByteType => exprBuilder.setByteVal(value.asInstanceOf[Byte].toInt)
+        case _: ShortType => exprBuilder.setShortVal(value.asInstanceOf[Short].toInt)
         case _: IntegerType | _: DateType => exprBuilder.setIntVal(value.asInstanceOf[Int])
         case _: LongType | _: TimestampType | _: TimestampNTZType =>
           exprBuilder.setLongVal(value.asInstanceOf[Long])
diff --git a/spark/src/main/scala/org/apache/spark/sql/comet/CometBatchScanExec.scala b/spark/src/main/scala/org/apache/spark/sql/comet/CometBatchScanExec.scala
@@ -66,7 +66,7 @@ case class CometBatchScanExec(wrapped: BatchScanExec, runtimeFilters: Seq[Expres
 
         override def next(): ColumnarBatch = {
           val batch = batches.next()
-          numOutputRows += batch.numRows()
+          numOutputRows += batch.numRows().toLong
           batch
         }
       }
diff --git a/spark/src/main/scala/org/apache/spark/sql/comet/CometBroadcastExchangeExec.scala b/spark/src/main/scala/org/apache/spark/sql/comet/CometBroadcastExchangeExec.scala
@@ -146,7 +146,7 @@ case class CometBroadcastExchangeExec(
         longMetric("numOutputRows") += numRows
         if (numRows >= maxBroadcastRows) {
           throw QueryExecutionErrors.cannotBroadcastTableOverMaxTableRowsError(
-            maxBroadcastRows,
+            maxBroadcastRows.toLong,
             numRows)
         }
 
diff --git a/spark/src/main/scala/org/apache/spark/sql/comet/CometCollectLimitExec.scala b/spark/src/main/scala/org/apache/spark/sql/comet/CometCollectLimitExec.scala
@@ -88,7 +88,7 @@ case class CometCollectLimitExec(
           outputPartitioning,
           serializer,
           metrics)
-        metrics("numPartitions").set(dep.partitioner.numPartitions)
+        metrics("numPartitions").set(dep.partitioner.numPartitions.toLong)
 
         new CometShuffledBatchRDD(dep, readMetrics)
       }
diff --git a/spark/src/main/scala/org/apache/spark/sql/comet/CometColumnarToRowExec.scala b/spark/src/main/scala/org/apache/spark/sql/comet/CometColumnarToRowExec.scala
@@ -82,7 +82,7 @@ case class CometColumnarToRowExec(child: SparkPlan)
       val toUnsafe = UnsafeProjection.create(localOutput, localOutput)
       batches.flatMap { batch =>
         numInputBatches += 1
-        numOutputRows += batch.numRows()
+        numOutputRows += batch.numRows().toLong
         batch.rowIterator().asScala.map(toUnsafe)
       }
     }
@@ -120,7 +120,7 @@ case class CometColumnarToRowExec(child: SparkPlan)
           .flatMap(CometUtils.decodeBatches(_, this.getClass.getSimpleName))
           .flatMap { batch =>
             numInputBatches += 1
-            numOutputRows += batch.numRows()
+            numOutputRows += batch.numRows().toLong
             batch.rowIterator().asScala.map(toUnsafe)
           }
 
diff --git a/spark/src/main/scala/org/apache/spark/sql/comet/CometScanExec.scala b/spark/src/main/scala/org/apache/spark/sql/comet/CometScanExec.scala
@@ -242,7 +242,7 @@ case class CometScanExec(
       driverMetrics("staticFilesSize") = filesSize
     }
     if (relation.partitionSchema.nonEmpty) {
-      driverMetrics("numPartitions") = partitions.length
+      driverMetrics("numPartitions") = partitions.length.toLong
     }
   }
 
@@ -284,7 +284,7 @@ case class CometScanExec(
 
         override def next(): ColumnarBatch = {
           val batch = batches.next()
-          numOutputRows += batch.numRows()
+          numOutputRows += batch.numRows().toLong
           batch
         }
       }
diff --git a/spark/src/main/scala/org/apache/spark/sql/comet/CometSparkToColumnarExec.scala b/spark/src/main/scala/org/apache/spark/sql/comet/CometSparkToColumnarExec.scala
@@ -83,7 +83,7 @@ case class CometSparkToColumnarExec(child: SparkPlan)
         val startNs = System.nanoTime()
         val batch = iter.next()
         conversionTime += System.nanoTime() - startNs
-        numInputRows += batch.numRows()
+        numInputRows += batch.numRows().toLong
         numOutputBatches += 1
         batch
       }
@@ -123,7 +123,7 @@ case class CometSparkToColumnarExec(child: SparkPlan)
             CometArrowConverters.rowToArrowBatchIter(
               sparkBatches,
               schema,
-              maxRecordsPerBatch,
+              maxRecordsPerBatch.toLong,
               timeZoneId,
               context)
           createTimingIter(arrowBatches, numInputRows, numOutputBatches, conversionTime)
diff --git a/spark/src/main/scala/org/apache/spark/sql/comet/CometTakeOrderedAndProjectExec.scala b/spark/src/main/scala/org/apache/spark/sql/comet/CometTakeOrderedAndProjectExec.scala
@@ -96,7 +96,7 @@ case class CometTakeOrderedAndProjectExec(
           outputPartitioning,
           serializer,
           metrics)
-        metrics("numPartitions").set(dep.partitioner.numPartitions)
+        metrics("numPartitions").set(dep.partitioner.numPartitions.toLong)
 
         new CometShuffledBatchRDD(dep, readMetrics)
       }
diff --git a/spark/src/main/scala/org/apache/spark/sql/comet/execution/shuffle/CometBlockStoreShuffleReader.scala b/spark/src/main/scala/org/apache/spark/sql/comet/execution/shuffle/CometBlockStoreShuffleReader.scala
@@ -111,7 +111,7 @@ class CometBlockStoreShuffleReader[K, C](
     // Update the context task metrics for each record read.
     val metricIter = CompletionIterator[(Any, Any), Iterator[(Any, Any)]](
       recordIter.map { record =>
-        readMetrics.incRecordsRead(record._2.numRows())
+        readMetrics.incRecordsRead(record._2.numRows().toLong)
         record
       },
       context.taskMetrics().mergeShuffleReadMetrics())
diff --git a/spark/src/main/scala/org/apache/spark/sql/comet/execution/shuffle/CometShuffleExchangeExec.scala b/spark/src/main/scala/org/apache/spark/sql/comet/execution/shuffle/CometShuffleExchangeExec.scala
@@ -137,7 +137,7 @@ case class CometShuffleExchangeExec(
         outputPartitioning,
         serializer,
         metrics)
-      metrics("numPartitions").set(dep.partitioner.numPartitions)
+      metrics("numPartitions").set(dep.partitioner.numPartitions.toLong)
       val executionId = sparkContext.getLocalProperty(SQLExecution.EXECUTION_ID_KEY)
       SQLMetrics.postDriverMetricUpdates(
         sparkContext,
@@ -151,7 +151,7 @@ case class CometShuffleExchangeExec(
         outputPartitioning,
         serializer,
         metrics)
-      metrics("numPartitions").set(dep.partitioner.numPartitions)
+      metrics("numPartitions").set(dep.partitioner.numPartitions.toLong)
       val executionId = sparkContext.getLocalProperty(SQLExecution.EXECUTION_ID_KEY)
       SQLMetrics.postDriverMetricUpdates(
         sparkContext,
@@ -385,7 +385,7 @@ object CometShuffleExchangeExec extends ShimCometShuffleExchangeExec {
         // end up being almost the same regardless of the index. substantially scrambling the
         // seed by hashing will help. Refer to SPARK-21782 for more details.
         val partitionId = TaskContext.get().partitionId()
-        var position = new XORShiftRandom(partitionId).nextInt(numPartitions)
+        var position = new XORShiftRandom(partitionId.toLong).nextInt(numPartitions)
         (_: InternalRow) => {
           // The HashPartitioner will handle the `mod` by the number of partitions
           position += 1
@@ -432,7 +432,7 @@ object CometShuffleExchangeExec extends ShimCometShuffleExchangeExec {
                 row: InternalRow): UnsafeExternalRowSorter.PrefixComputer.Prefix = {
               // The hashcode generated from the binary form of a [[UnsafeRow]] should not be null.
               result.isNull = false
-              result.value = row.hashCode()
+              result.value = row.hashCode().toLong
               result
             }
           }

Original file line number	Diff line number	Diff line change
`@@ -66,7 +66,7 @@ case class CometBatchScanExec(wrapped: BatchScanExec, runtimeFilters: Seq[Expres`
`66`	`66`
`67`	`67`	`override def next(): ColumnarBatch = {`
`68`	`68`	`val batch = batches.next()`
`69`		`- numOutputRows += batch.numRows()`
	`69`	`+ numOutputRows += batch.numRows().toLong`
`70`	`70`	`batch`
`71`	`71`	`}`
`72`	`72`	`}`
Original file line number	Diff line number	Diff line change
`@@ -146,7 +146,7 @@ case class CometBroadcastExchangeExec(`
`146`	`146`	`longMetric("numOutputRows") += numRows`
`147`	`147`	`if (numRows >= maxBroadcastRows) {`
`148`	`148`	`throw QueryExecutionErrors.cannotBroadcastTableOverMaxTableRowsError(`
`149`		`- maxBroadcastRows,`
	`149`	`+ maxBroadcastRows.toLong,`
`150`	`150`	`numRows)`
`151`	`151`	`}`
`152`	`152`
Original file line number	Diff line number	Diff line change
`@@ -88,7 +88,7 @@ case class CometCollectLimitExec(`
`88`	`88`	`outputPartitioning,`
`89`	`89`	`serializer,`
`90`	`90`	`metrics)`
`91`		`- metrics("numPartitions").set(dep.partitioner.numPartitions)`
	`91`	`+ metrics("numPartitions").set(dep.partitioner.numPartitions.toLong)`
`92`	`92`
`93`	`93`	`new CometShuffledBatchRDD(dep, readMetrics)`
`94`	`94`	`}`
Original file line number	Diff line number	Diff line change
`@@ -82,7 +82,7 @@ case class CometColumnarToRowExec(child: SparkPlan)`
`82`	`82`	`val toUnsafe = UnsafeProjection.create(localOutput, localOutput)`
`83`	`83`	`batches.flatMap { batch =>`
`84`	`84`	`numInputBatches += 1`
`85`		`- numOutputRows += batch.numRows()`
	`85`	`+ numOutputRows += batch.numRows().toLong`
`86`	`86`	`batch.rowIterator().asScala.map(toUnsafe)`
`87`	`87`	`}`
`88`	`88`	`}`
`@@ -120,7 +120,7 @@ case class CometColumnarToRowExec(child: SparkPlan)`
`120`	`120`	`.flatMap(CometUtils.decodeBatches(_, this.getClass.getSimpleName))`
`121`	`121`	`.flatMap { batch =>`
`122`	`122`	`numInputBatches += 1`
`123`		`- numOutputRows += batch.numRows()`
	`123`	`+ numOutputRows += batch.numRows().toLong`
`124`	`124`	`batch.rowIterator().asScala.map(toUnsafe)`
`125`	`125`	`}`
`126`	`126`
Original file line number	Diff line number	Diff line change
`@@ -242,7 +242,7 @@ case class CometScanExec(`
`242`	`242`	`driverMetrics("staticFilesSize") = filesSize`
`243`	`243`	`}`
`244`	`244`	`if (relation.partitionSchema.nonEmpty) {`
`245`		`- driverMetrics("numPartitions") = partitions.length`
	`245`	`+ driverMetrics("numPartitions") = partitions.length.toLong`
`246`	`246`	`}`
`247`	`247`	`}`
`248`	`248`
`@@ -284,7 +284,7 @@ case class CometScanExec(`
`284`	`284`
`285`	`285`	`override def next(): ColumnarBatch = {`
`286`	`286`	`val batch = batches.next()`
`287`		`- numOutputRows += batch.numRows()`
	`287`	`+ numOutputRows += batch.numRows().toLong`
`288`	`288`	`batch`
`289`	`289`	`}`
`290`	`290`	`}`
Original file line number	Diff line number	Diff line change
`@@ -96,7 +96,7 @@ case class CometTakeOrderedAndProjectExec(`
`96`	`96`	`outputPartitioning,`
`97`	`97`	`serializer,`
`98`	`98`	`metrics)`
`99`		`- metrics("numPartitions").set(dep.partitioner.numPartitions)`
	`99`	`+ metrics("numPartitions").set(dep.partitioner.numPartitions.toLong)`
`100`	`100`
`101`	`101`	`new CometShuffledBatchRDD(dep, readMetrics)`
`102`	`102`	`}`