Revert "Spark 3.4: Support distribute by any predefined transform"

This reverts commit 93a2b9d.

Author: pan3793

spark-3.4/clickhouse-spark-it/src/test/scala/org/apache/spark/sql/clickhouse/cluster/ClusterShardByTransformSuite.scala

@@ -78,8 +78,12 @@ class WriteDistributionAndOrderingSuite extends SparkClickHouseSingleTest {
     WRITE_REPARTITION_BY_PARTITION.key -> repartitionByPartition.toString,
     WRITE_LOCAL_SORT_BY_KEY.key -> localSortByKey.toString
   ) {
-    write()
-    check()
+    if (!ignoreUnsupportedTransform && repartitionByPartition) {
+      intercept[AnalysisException](write())
+    } else {
+      write()
+      check()
+    }
   }
 
   Seq(true, false).foreach { ignoreUnsupportedTransform =>

spark-3.4/clickhouse-spark-it/src/test/scala/org/apache/spark/sql/clickhouse/single/WriteDistributionAndOrderingSuite.scala

@@ -16,21 +16,18 @@ package org.apache.spark.sql.clickhouse
 
 import org.apache.spark.sql.AnalysisException
 import org.apache.spark.sql.catalyst.SQLConfHelper
-import org.apache.spark.sql.catalyst.analysis.NoSuchFunctionException
-import org.apache.spark.sql.catalyst.expressions.{BoundReference, Expression, TransformExpression}
+import org.apache.spark.sql.catalyst.expressions.{BoundReference, Expression}
 import org.apache.spark.sql.clickhouse.ClickHouseSQLConf.IGNORE_UNSUPPORTED_TRANSFORM
-import org.apache.spark.sql.connector.catalog.Identifier
-import org.apache.spark.sql.connector.catalog.functions.{BoundFunction, ScalarFunction, UnboundFunction}
 import org.apache.spark.sql.connector.expressions.Expressions._
 import org.apache.spark.sql.connector.expressions.{Expression => V2Expression, _}
 import org.apache.spark.sql.types.{IntegerType, LongType, StructField, StructType}
 import xenon.clickhouse.exception.CHClientException
 import xenon.clickhouse.expr._
-import xenon.clickhouse.func.FunctionRegistry
 
+import scala.annotation.tailrec
 import scala.util.{Failure, Success, Try}
 
-class ExprUtils(functionRegistry: FunctionRegistry) extends SQLConfHelper with Serializable {
+object ExprUtils extends SQLConfHelper {
 
   def toSparkPartitions(partitionKey: Option[List[Expr]]): Array[Transform] =
     partitionKey.seq.flatten.flatten(toSparkTransformOpt).toArray
@@ -50,28 +47,7 @@ class ExprUtils(functionRegistry: FunctionRegistry) extends SQLConfHelper with S
         toSparkTransformOpt(expr).map(trans => Expressions.sort(trans, direction, nullOrder))
       }.toArray
 
-  private def loadV2FunctionOpt(
-    name: String,
-    args: Seq[Expression]
-  ): Option[BoundFunction] = {
-    def loadFunction(ident: Identifier): UnboundFunction =
-      functionRegistry.load(ident.name).getOrElse(throw new NoSuchFunctionException(ident))
-    val inputType = StructType(args.zipWithIndex.map {
-      case (exp, pos) => StructField(s"_$pos", exp.dataType, exp.nullable)
-    })
-    try {
-      val unbound = loadFunction(Identifier.of(Array.empty, name))
-      Some(unbound.bind(inputType))
-    } catch {
-      case e: NoSuchFunctionException =>
-        throw e
-      case _: UnsupportedOperationException if conf.getConf(IGNORE_UNSUPPORTED_TRANSFORM) =>
-        None
-      case e: UnsupportedOperationException =>
-        throw new AnalysisException(e.getMessage, cause = Some(e))
-    }
-  }
-
+  @tailrec
   def toCatalyst(v2Expr: V2Expression, fields: Array[StructField]): Expression =
     v2Expr match {
       case IdentityTransform(ref) => toCatalyst(ref, fields)
@@ -81,15 +57,8 @@ class ExprUtils(functionRegistry: FunctionRegistry) extends SQLConfHelper with S
           .find { case (field, _) => field.name == ref.fieldNames.head }
           .getOrElse(throw CHClientException(s"Invalid field reference: $ref"))
         BoundReference(ordinal, field.dataType, field.nullable)
-      case t: Transform =>
-        val catalystArgs = t.arguments().map(toCatalyst(_, fields))
-        loadV2FunctionOpt(t.name(), catalystArgs).map { bound =>
-          TransformExpression(bound, catalystArgs)
-        }.getOrElse {
-          throw CHClientException(s"Unsupported expression: $v2Expr")
-        }
       case _ => throw CHClientException(
-          s"Unsupported expression: $v2Expr"
+          s"Unsupported V2 expression: $v2Expr, SPARK-33779: Spark 3.3 only support IdentityTransform"
         )
     }
 
@@ -114,10 +83,10 @@ class ExprUtils(functionRegistry: FunctionRegistry) extends SQLConfHelper with S
     case FuncExpr("toYYYYMMDD", List(FieldRef(col))) => days(col)
     case FuncExpr("toHour", List(FieldRef(col))) => hours(col)
     case FuncExpr("HOUR", List(FieldRef(col))) => hours(col)
+    // TODO support arbitrary functions
+    // case FuncExpr("xxHash64", List(FieldRef(col))) => apply("ck_xx_hash64", column(col))
     case FuncExpr("rand", Nil) => apply("rand")
     case FuncExpr("toYYYYMMDD", List(FuncExpr("toDate", List(FieldRef(col))))) => identity(col)
-    case FuncExpr(funName, List(FieldRef(col))) if functionRegistry.clickHouseToSparkFunc.contains(funName) =>
-      apply(functionRegistry.clickHouseToSparkFunc(funName), column(col))
     case unsupported => throw CHClientException(s"Unsupported ClickHouse expression: $unsupported")
   }
 
@@ -127,8 +96,7 @@ class ExprUtils(functionRegistry: FunctionRegistry) extends SQLConfHelper with S
     case DaysTransform(FieldReference(Seq(col))) => FuncExpr("toYYYYMMDD", List(FieldRef(col)))
     case HoursTransform(FieldReference(Seq(col))) => FuncExpr("toHour", List(FieldRef(col)))
     case IdentityTransform(fieldRefs) => FieldRef(fieldRefs.describe)
-    case ApplyTransform(name, args) if functionRegistry.sparkToClickHouseFunc.contains(name) =>
-      FuncExpr(functionRegistry.sparkToClickHouseFunc(name), args.map(arg => SQLExpr(arg.describe())).toList)
+    case ApplyTransform(name, args) => FuncExpr(name, args.map(arg => SQLExpr(arg.describe())).toList)
     case bucket: BucketTransform => throw CHClientException(s"Bucket transform not support yet: $bucket")
     case other: Transform => throw CHClientException(s"Unsupported transform: $other")
   }
@@ -145,18 +113,8 @@ class ExprUtils(functionRegistry: FunctionRegistry) extends SQLConfHelper with S
     case IdentityTransform(FieldReference(Seq(col))) => primarySchema.find(_.name == col)
         .orElse(secondarySchema.find(_.name == col))
         .getOrElse(throw CHClientException(s"Invalid partition column: $col"))
-    case t @ ApplyTransform(transformName, _) =>
-      val resType =
-        functionRegistry.load(transformName).getOrElse(throw new NoSuchFunctionException(transformName)) match {
-          case f: ScalarFunction[_] => f.resultType()
-          case other => throw CHClientException(s"Unsupported function: $other")
-        }
-      StructField(t.toString, resType)
+    case ckXxhHash64 @ ApplyTransform("ck_xx_hash64", _) => StructField(ckXxhHash64.toString, LongType)
     case bucket: BucketTransform => throw CHClientException(s"Bucket transform not support yet: $bucket")
     case other: Transform => throw CHClientException(s"Unsupported transform: $other")
   }
 }
-
-object ExprUtils {
-  def apply(functionRegistry: FunctionRegistry): ExprUtils = new ExprUtils(functionRegistry)
-}

spark-3.4/clickhouse-spark/src/main/scala/org/apache/spark/sql/clickhouse/ExprUtils.scala

@@ -85,15 +85,12 @@ class ClickHouseCatalog extends TableCatalog
 
     val dynamicFunctionRegistry = new DynamicFunctionRegistry
     val xxHash64ShardFunc = new ClickHouseXxHash64Shard(clusterSpecs)
-    val monthsFunc = new Months()
     dynamicFunctionRegistry.register("ck_xx_hash64_shard", xxHash64ShardFunc) // for compatible
     dynamicFunctionRegistry.register("clickhouse_shard_xxHash64", xxHash64ShardFunc)
-    dynamicFunctionRegistry.register("months", monthsFunc)
     this.functionRegistry = new CompositeFunctionRegistry(Array(StaticFunctionRegistry, dynamicFunctionRegistry))
 
     log.info(s"Detect ${clusterSpecs.size} ClickHouse clusters: ${clusterSpecs.map(_.name).mkString(",")}")
     log.info(s"ClickHouse clusters' detail: $clusterSpecs")
-    log.info(s"functionRegistry: ${this.functionRegistry.list.mkString(",")}")
   }
 
   override def name(): String = catalogName
@@ -144,8 +141,7 @@ class ClickHouseCatalog extends TableCatalog
       tableClusterSpec,
       _tz,
       tableSpec,
-      tableEngineSpec,
-      functionRegistry
+      tableEngineSpec
     )
   }
 
@@ -210,7 +206,7 @@ class ClickHouseCatalog extends TableCatalog
 
     val partitionsClause = partitions match {
       case transforms if transforms.nonEmpty =>
-        transforms.map(ExprUtils(functionRegistry).toClickHouse(_).sql).mkString("PARTITION BY (", ", ", ")")
+        transforms.map(ExprUtils.toClickHouse(_).sql).mkString("PARTITION BY (", ", ", ")")
       case _ => ""
     }
 
@@ -301,7 +297,7 @@ class ClickHouseCatalog extends TableCatalog
       }
     tableOpt match {
       case None => false
-      case Some(ClickHouseTable(_, cluster, _, tableSpec, _, _)) =>
+      case Some(ClickHouseTable(_, cluster, _, tableSpec, _)) =>
         val (db, tbl) = (tableSpec.database, tableSpec.name)
         val isAtomic = loadNamespaceMetadata(Array(db)).get("engine").equalsIgnoreCase("atomic")
         val syncClause = if (isAtomic) "SYNC" else ""

spark-3.4/clickhouse-spark/src/main/scala/xenon/clickhouse/ClickHouseCatalog.scala

@@ -14,12 +14,16 @@
 
 package xenon.clickhouse
 
-import org.apache.spark.sql.catalyst.expressions.GenericInternalRow
+import java.lang.{Integer => JInt, Long => JLong}
+import java.time.{LocalDate, ZoneId}
+import java.util
+import scala.collection.JavaConverters._
 import org.apache.spark.sql.catalyst.{InternalRow, SQLConfHelper}
-import org.apache.spark.sql.clickhouse.ClickHouseSQLConf.{READ_DISTRIBUTED_CONVERT_LOCAL, USE_NULLABLE_QUERY_SCHEMA}
+import org.apache.spark.sql.catalyst.expressions.GenericInternalRow
 import org.apache.spark.sql.clickhouse.{ExprUtils, ReadOptions, WriteOptions}
-import org.apache.spark.sql.connector.catalog.TableCapability._
+import org.apache.spark.sql.clickhouse.ClickHouseSQLConf.{READ_DISTRIBUTED_CONVERT_LOCAL, USE_NULLABLE_QUERY_SCHEMA}
 import org.apache.spark.sql.connector.catalog._
+import org.apache.spark.sql.connector.catalog.TableCapability._
 import org.apache.spark.sql.connector.expressions.Transform
 import org.apache.spark.sql.connector.read.ScanBuilder
 import org.apache.spark.sql.connector.write.LogicalWriteInfo
@@ -30,23 +34,16 @@ import org.apache.spark.unsafe.types.UTF8String
 import xenon.clickhouse.Utils._
 import xenon.clickhouse.client.NodeClient
 import xenon.clickhouse.expr.{Expr, OrderExpr}
-import xenon.clickhouse.func.FunctionRegistry
 import xenon.clickhouse.read.{ClickHouseMetadataColumn, ClickHouseScanBuilder, ScanJobDescription}
 import xenon.clickhouse.spec._
 import xenon.clickhouse.write.{ClickHouseWriteBuilder, WriteJobDescription}
 
-import java.lang.{Integer => JInt, Long => JLong}
-import java.time.{LocalDate, ZoneId}
-import java.util
-import scala.collection.JavaConverters._
-
 case class ClickHouseTable(
   node: NodeSpec,
   cluster: Option[ClusterSpec],
   implicit val tz: ZoneId,
   spec: TableSpec,
-  engineSpec: TableEngineSpec,
-  functionRegistry: FunctionRegistry
+  engineSpec: TableEngineSpec
 ) extends Table
     with SupportsRead
     with SupportsWrite
@@ -133,12 +130,10 @@ case class ClickHouseTable(
   private lazy val metadataSchema: StructType =
     StructType(metadataColumns.map(_.asInstanceOf[ClickHouseMetadataColumn].toStructField))
 
-  override lazy val partitioning: Array[Transform] = ExprUtils(functionRegistry).toSparkPartitions(partitionKey)
+  override lazy val partitioning: Array[Transform] = ExprUtils.toSparkPartitions(partitionKey)
 
   override lazy val partitionSchema: StructType = StructType(
-    partitioning.map(partTransform =>
-      ExprUtils(functionRegistry).inferTransformSchema(schema, metadataSchema, partTransform)
-    )
+    partitioning.map(partTransform => ExprUtils.inferTransformSchema(schema, metadataSchema, partTransform))
   )
 
   override lazy val properties: util.Map[String, String] = spec.toJavaMap
@@ -175,8 +170,7 @@ case class ClickHouseTable(
       shardingKey = shardingKey,
       partitionKey = partitionKey,
       sortingKey = sortingKey,
-      writeOptions = new WriteOptions(info.options.asCaseSensitiveMap()),
-      functionRegistry = functionRegistry
+      writeOptions = new WriteOptions(info.options.asCaseSensitiveMap())
     )
 
     new ClickHouseWriteBuilder(writeJob)