Блог

Я пытаюсь преобразовать SQL-запрос в выражение реляционной алгебры, используя Apache Calcite SqlToRelConverter .

Для этого запроса все работает нормально (кавычки указаны в нижнем регистре):

 queryToRelationalAlgebraRoot("SELECT "country" FROM "mytable"")
  

Но в этом запросе происходит сбой:

 queryToRelationalAlgebraRoot("SELECT "country", SUM("salary") FROM "mytable" GROUP BY "country"")
  

с этой ошибкой:

org.apache.calcite.sql.validate.SqlValidatorException: No match found for function signature SUM(<NUMERIC>)

Похоже, что каким-то образом в средстве проверки SQL не зарегистрированы функции агрегирования, такие как sum или count.

 case class Income(id: Int, salary: Double, country: String)

class SparkDataFrameTable(df: DataFrame) extends AbstractTable {

  def getRowType(typeFactory: RelDataTypeFactory): RelDataType = {
    val typeList = df.schema.fields.map {
      field => field.dataType match {
        case t: StringType => typeFactory.createSqlType(SqlTypeName.VARCHAR)
        case t: IntegerType => typeFactory.createSqlType(SqlTypeName.INTEGER)
        case t: DoubleType => typeFactory.createSqlType(SqlTypeName.DOUBLE)
      }
    }.toList.asJava
    val fieldNameList = df.schema.fieldNames.toList.asJava
    typeFactory.createStructType(typeList, fieldNameList)
  }

}

object RelationalAlgebra {

  def queryToRelationalAlgebraRoot(query: String): RelRoot = {
    val sqlParser = SqlParser.create(query)
    val sqlParseTree = sqlParser.parseQuery()

    val frameworkConfig = Frameworks.newConfigBuilder().build()
    val planner = new PlannerImpl(frameworkConfig)

    val rootSchema = CalciteSchema.createRootSchema(true, true)

    // some sample data for testing
    val inc1 = new Income(1, 100000, "USA")
    val inc2 = new Income(2, 110000, "USA")
    val inc3 = new Income(3, 80000, "Canada")
    val spark = SparkSession.builder().master("local").getOrCreate()
    import spark.implicits._
    val df = Seq(inc1, inc2, inc3).toDF()
    rootSchema.add("mytable", new SparkDataFrameTable(df))

    val defaultSchema = List[String]().asJava
    val calciteConnectionConfigProperties = new Properties()
    val calciteConnectionConfigImpl = new CalciteConnectionConfigImpl(calciteConnectionConfigProperties)
    val sqlTypeFactoryImpl = new SqlTypeFactoryImpl(RelDataTypeSystem.DEFAULT)
    val calciteCatelogReader = new CalciteCatalogReader(rootSchema, defaultSchema, sqlTypeFactoryImpl, calciteConnectionConfigImpl)
    val defaultValidator = SqlValidatorUtil.newValidator(new SqlStdOperatorTable(), calciteCatelogReader, sqlTypeFactoryImpl, SqlConformanceEnum.LENIENT)

    val relExpressionOptimizationCluster = RelOptCluster.create(new VolcanoPlanner(), new RexBuilder(sqlTypeFactoryImpl))

    val sqlToRelConfig = SqlToRelConverter.configBuilder().build()

    val sqlToRelConverter = new SqlToRelConverter(planner, defaultValidator, calciteCatelogReader, relExpressionOptimizationCluster, StandardConvertletTable.INSTANCE, sqlToRelConfig)

    sqlToRelConverter.convertQuery(sqlParseTree, true, true)
  }

}
  

Проблема с кодом заключается в том, что new SqlStdOperatorTable() создается средство проверки, которое не инициализировано. Правильный способ использования SqlStdOperatorTable — использовать SqlStdOperatorTable.instance() .

Я нашел решение после отправки электронной почты dev@calcite.apache.org список рассылки. Я хотел бы поблагодарить Ючжао Чена за то, что он изучил мой вопрос и указал на проблему с моим кодом.

Я не знаком с api, но вашему SQL нужна группа по стране. И если инструмент должен получить этот вывод и использовать его, вероятно, потребуется, чтобы вы также назвали столбец псевдонимом.