peterxcli commented on issue #4724: URL: https://github.com/apache/datafusion-comet/issues/4724#issuecomment-5242773261
> above Issue reproducible with v0.17.0 using below sample queries and data. > > `SELECT imp_campaign_id, imp_adgroup_id, ARRAY_JOIN(COLLECT_SET(STRING(imp_creative_id)), ',') AS creatives, COUNT(DISTINCT imp_impression_id) AS imp_count FROM impressions GROUP BY imp_campaign_id, imp_adgroup_id` > > `SELECT imp_campaign_id, COUNT(DISTINCT imp_user_id) AS users, COLLECT_SET(imp_creative_id) AS creatives FROM impressions GROUP BY imp_campaign_id` > > [impressions.csv](https://github.com/user-attachments/files/30888137/impressions.csv) https://github.com/apache/datafusion-comet/pull/4727 should fix this. you can 1. checkout to that pr and run ```bash make release PROFILES="-Pspark-4.1 -Drat.skip=true" export COMET_JAR=spark/target/comet-spark-spark4.1_2.13-1.0.0-SNAPSHOT.jar sdk use spark 4.1.0 ``` 2. store following scala script as `4724.scala` 3. download csv with `curl -L https://github.com/user-attachments/files/30888137/impressions.csv -o impressions.csv` 5. start spark shell ```bash spark-shell \ --jars $COMET_JAR \ --conf spark.driver.extraClassPath=$COMET_JAR \ --conf spark.executor.extraClassPath=$COMET_JAR \ --conf spark.plugins=org.apache.spark.CometPlugin \ --conf spark.shuffle.manager=org.apache.spark.sql.comet.execution.shuffle.CometShuffleManager \ --conf spark.comet.explainFallback.enabled=true \ --conf spark.memory.offHeap.enabled=true \ --conf spark.memory.offHeap.size=4g ``` 6. in spark shell, run `:load 4724.scala` `4724.scala`: ```scala val input = "impressions.csv" val parquet = "impressions.parquet" spark.read .option("header", "true") .option("inferSchema", "true") .csv(input) .repartition(2) .write .mode("overwrite") .parquet(parquet) spark.read.parquet(parquet).createOrReplaceTempView("impressions") val q1 = """ |SELECT | imp_campaign_id, | imp_adgroup_id, | ARRAY_JOIN(COLLECT_SET(STRING(imp_creative_id)), ',') AS creatives, | COUNT(DISTINCT imp_impression_id) AS imp_count |FROM impressions |GROUP BY imp_campaign_id, imp_adgroup_id |""".stripMargin val q2 = """ |SELECT | imp_campaign_id, | COUNT(DISTINCT imp_user_id) AS users, | COLLECT_SET(imp_creative_id) AS creatives |FROM impressions |GROUP BY imp_campaign_id |""".stripMargin def run(comet: Boolean) = { spark.conf.set("spark.comet.enabled", comet.toString) val df1 = spark.sql(q1) val result1 = df1.collect().map { r => ( r.getInt(0), r.getInt(1), r.getString(2).split(",").toSet, r.getLong(3)) }.toSet val df2 = spark.sql(q2) val result2 = df2.collect().map { r => (r.getInt(0), r.getLong(1), r.getSeq[Int](2).toSet) }.toSet (result1, result2, Seq( df1.queryExecution.executedPlan.toString, df2.queryExecution.executedPlan.toString)) } val sparkResult = run(false) val cometResult = run(true) val answersMatch = ( sparkResult._1 == cometResult._1 && sparkResult._2 == cometResult._2 ) val requiredPlanParts = Seq( "CometNativeScan", "CometExchange", "CometHashAggregate", "[Partial]", "[PartialMerge]", "[PartialMerge, Partial]", "[Final]", "merge_collect_set") val fullyNative = cometResult._3.forall(plan => requiredPlanParts.forall(plan.contains)) println(s"answersMatch=$answersMatch fullyNative=$fullyNative") cometResult._3.foreach(println) if (!answersMatch || !fullyNative) System.exit(1) System.exit(0) ``` -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
