vikramahuja1001 commented on a change in pull request #4141:
URL: https://github.com/apache/carbondata/pull/4141#discussion_r655281093
##########
File path:
integration/spark/src/test/scala/org/apache/carbondata/spark/testsuite/index/TestIndexCommand.scala
##########
@@ -39,7 +39,9 @@ class TestIndexCommand extends QueryTest with
BeforeAndAfterAll {
val newClass = "org.apache.spark.sql.CarbonSource"
- test("test index create: don't support using non-exist class") {
+ // TODO: This issue will be fixed during alter change column support for
spark 3.1.1
+ // REFER: https://issues.apache.org/jira/browse/CARBONDATA-4210
+ ignore("test index create: don't support using non-exist class") {
Review comment:
changed these test cases back, and changed the expected error message
##########
File path:
integration/spark/src/test/scala/org/apache/spark/carbondata/bucketing/TableBucketingTestCase.scala
##########
@@ -38,13 +38,13 @@ import
org.apache.carbondata.spark.exception.ProcessMetaDataException
class TableBucketingTestCase extends QueryTest with BeforeAndAfterAll {
- var threshold: Int = _
+ var threshold: String = _
override def beforeAll {
CarbonProperties.getInstance()
.addProperty(CarbonCommonConstants.CARBON_TIMESTAMP_FORMAT, "yyyy/MM/dd")
- threshold =
sqlContext.getConf("spark.sql.autoBroadcastJoinThreshold").toInt
+ threshold = sqlContext.getConf("spark.sql.autoBroadcastJoinThreshold")
Review comment:
changed
##########
File path:
examples/spark/src/main/scala/org/apache/carbondata/examples/LuceneIndexExample.scala
##########
@@ -61,37 +61,37 @@ object LuceneIndexExample {
""".stripMargin)
// 1. Compare the performance:
-
- def time(code: => Unit): Double = {
- val start = System.currentTimeMillis()
- code
- // return time in second
- (System.currentTimeMillis() - start).toDouble / 1000
- }
-
- val timeWithoutLuceneIndex = time {
-
- spark.sql(
- s"""
- | SELECT count(*)
- | FROM personTable where id like '% test1 %'
- """.stripMargin).show()
-
- }
-
- val timeWithLuceneIndex = time {
-
- spark.sql(
- s"""
- | SELECT count(*)
- | FROM personTable where TEXT_MATCH('id:test1')
- """.stripMargin).show()
-
- }
+ // TODO: Revert this after SPARK 3.1 fix
Review comment:
changed this
##########
File path:
integration/spark/src/main/scala/org/apache/carbondata/spark/load/CsvRDDHelper.scala
##########
@@ -26,7 +26,7 @@ import org.apache.hadoop.mapreduce.{TaskAttemptID, TaskType}
import org.apache.hadoop.mapreduce.lib.input.{FileInputFormat, FileSplit}
import org.apache.hadoop.mapreduce.task.TaskAttemptContextImpl
import org.apache.spark.rdd.RDD
-import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.{CarbonToSparkAdapter, SparkSession}
Review comment:
That old package has been renamed
##########
File path:
examples/spark/src/main/scala/org/apache/carbondata/examples/LuceneIndexExample.scala
##########
@@ -60,40 +60,6 @@ object LuceneIndexExample {
| AS 'lucene'
""".stripMargin)
- // 1. Compare the performance:
-
- def time(code: => Unit): Double = {
- val start = System.currentTimeMillis()
- code
- // return time in second
- (System.currentTimeMillis() - start).toDouble / 1000
- }
-
- val timeWithoutLuceneIndex = time {
-
- spark.sql(
- s"""
- | SELECT count(*)
- | FROM personTable where id like '% test1 %'
- """.stripMargin).show()
-
- }
-
- val timeWithLuceneIndex = time {
-
- spark.sql(
Review comment:
done
##########
File path:
integration/spark/src/main/spark3.1/org/apache/spark/sql/CarbonToSparkAdapter.scala
##########
@@ -0,0 +1,735 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql
+
+import java.net.URI
+import java.sql.{Date, Timestamp}
+import java.time.ZoneId
+
+import javax.xml.bind.DatatypeConverter
+
+import scala.annotation.tailrec
+import scala.collection.mutable
+import scala.collection.mutable.ArrayBuffer
+
+import org.antlr.v4.runtime.tree.TerminalNode
+import org.apache.spark.{SparkContext, TaskContext}
+import org.apache.spark.rdd.RDD
+import org.apache.spark.scheduler.{SparkListener, SparkListenerApplicationEnd}
+import org.apache.spark.serializer.Serializer
+import
org.apache.spark.sql.carbondata.execution.datasources.CarbonFileIndexReplaceRule
+import org.apache.spark.sql.catalyst.{CarbonParserUtil, InternalRow,
QueryPlanningTracker, TableIdentifier}
+import org.apache.spark.sql.catalyst.analysis.{Analyzer, UnresolvedRelation}
+import org.apache.spark.sql.catalyst.catalog.{CatalogStorageFormat,
ExternalCatalogWithListener, SessionCatalog}
+import org.apache.spark.sql.catalyst.encoders.RowEncoder
+import org.apache.spark.sql.catalyst.expressions.{Alias, Attribute,
AttributeMap, AttributeReference, AttributeSeq, AttributeSet, ExprId,
Expression, ExpressionSet, NamedExpression, Predicate, ScalaUDF, SortOrder,
SubqueryExpression}
+import org.apache.spark.sql.catalyst.expressions.codegen._
+import org.apache.spark.sql.catalyst.expressions.codegen.Block._
+import org.apache.spark.sql.catalyst.optimizer.{BuildLeft, BuildRight,
BuildSide, Optimizer}
+import org.apache.spark.sql.catalyst.parser.ParserUtils.operationNotAllowed
+import org.apache.spark.sql.catalyst.parser.SqlBaseParser.{BucketSpecContext,
ColTypeListContext, CreateTableHeaderContext, LocationSpecContext,
PartitionFieldListContext, QueryContext, SkewSpecContext,
TablePropertyListContext}
+import org.apache.spark.sql.catalyst.plans.{JoinType, QueryPlan, logical}
+import org.apache.spark.sql.catalyst.plans.logical.{ColumnStat,
InsertIntoStatement, Join, JoinHint, LogicalPlan, OneRowRelation,
QualifiedColType, Statistics, SubqueryAlias}
+import org.apache.spark.sql.catalyst.plans.physical.SinglePartition
+import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.catalyst.trees.{CurrentOrigin, TreeNode}
+import org.apache.spark.sql.catalyst.util.{DateTimeUtils, TimestampFormatter}
+import org.apache.spark.sql.execution.{ExplainMode, QueryExecution,
SQLExecution, ShuffledRowRDD, SimpleMode, SparkPlan, UnaryExecNode}
+import org.apache.spark.sql.execution.command.{ExplainCommand, Field,
PartitionerField, RefreshTableCommand, TableModel, TableNewProcessor}
+import
org.apache.spark.sql.execution.command.table.{CarbonCreateTableAsSelectCommand,
CarbonCreateTableCommand}
+import org.apache.spark.sql.execution.datasources.{DataSourceStrategy,
FilePartition, PartitionedFile}
+import org.apache.spark.sql.execution.exchange.ShuffleExchangeExec
+import org.apache.spark.sql.execution.metric.SQLShuffleWriteMetricsReporter
+import org.apache.spark.sql.execution.strategy.CarbonDataSourceScan
+import org.apache.spark.sql.hive.HiveExternalCatalog
+import org.apache.spark.sql.internal.{SessionState, SharedState}
+import org.apache.spark.sql.optimizer.{CarbonIUDRule, CarbonUDFTransformRule,
MVRewriteRule}
+import org.apache.spark.sql.parser.CarbonSpark2SqlParser
+import
org.apache.spark.sql.parser.CarbonSparkSqlParserUtil.{checkIfDuplicateColumnExists,
convertDbNameToLowerCase, validateStreamingProperty}
+import org.apache.spark.sql.secondaryindex.optimizer.CarbonSITransformationRule
+import org.apache.spark.sql.sources.Filter
+import org.apache.spark.sql.streaming.Trigger
+import org.apache.spark.sql.types.{AbstractDataType, CharType, DataType,
Metadata, StringType, StructField, VarcharType}
+import org.apache.spark.sql.util.SparkSQLUtil
+import org.apache.spark.unsafe.types.UTF8String
+
+import org.apache.carbondata.common.exceptions.DeprecatedFeatureException
+import
org.apache.carbondata.common.exceptions.sql.MalformedCarbonCommandException
+import org.apache.carbondata.core.constants.CarbonCommonConstants
+import org.apache.carbondata.core.datastore.impl.FileFactory
+import org.apache.carbondata.core.metadata.AbsoluteTableIdentifier
+import org.apache.carbondata.core.metadata.datatype.DataTypes
+import org.apache.carbondata.core.metadata.schema.SchemaReader
+import org.apache.carbondata.core.util.{CarbonProperties,
ThreadLocalSessionInfo}
+import org.apache.carbondata.core.util.path.CarbonTablePath
+import org.apache.carbondata.geo.{InPolygonJoinUDF, ToRangeListAsStringUDF}
+import org.apache.carbondata.mv.plans.modular.{GroupBy, ModularPlan, Select}
+import org.apache.carbondata.spark.CarbonOption
+import org.apache.carbondata.spark.util.CarbonScalaUtil
+
+object CarbonToSparkAdapter {
Review comment:
done
##########
File path:
integration/spark/src/main/scala/org/apache/spark/sql/catalyst/CarbonDDLSqlParser.scala
##########
@@ -65,7 +65,10 @@ abstract class CarbonDDLSqlParser extends
AbstractCarbonSparkSQLParser {
protected val ESCAPECHAR = carbonKeyWord("ESCAPECHAR")
protected val EXCLUDE = carbonKeyWord("EXCLUDE")
protected val EXPLAIN = carbonKeyWord("EXPLAIN")
- protected val EXTENDED = carbonKeyWord("EXTENDED")
+ protected val MODE = carbonKeyWord("EXTENDED") |
+ carbonKeyWord("CODEGEN") |
+ carbonKeyWord("COST") |
+ carbonKeyWord("FORMATTED")
Review comment:
JIRA: https://issues.apache.org/jira/browse/CARBONDATA-4201
Done
##########
File path:
integration/spark/src/main/scala/org/apache/spark/sql/execution/joins/BroadCastPolygonFilterPushJoin.scala
##########
@@ -47,16 +45,13 @@ case class BroadCastPolygonFilterPushJoin(
leftKeys: Seq[Expression],
rightKeys: Seq[Expression],
joinType: JoinType,
- buildSide: BuildSide,
condition: Option[Expression],
left: SparkPlan,
right: SparkPlan
-) extends BinaryExecNode with HashJoin {
Review comment:
methods overridden from HashJoin needed to be removed, so this itself
was not needed.
##########
File path:
processing/src/main/java/org/apache/carbondata/processing/loading/parser/impl/RowParserImpl.java
##########
@@ -109,7 +109,9 @@ public RowParserImpl(DataField[] output,
CarbonDataLoadConfiguration configurati
return new String[numberOfColumns];
}
// If number of columns are less in a row then create new array with same
size of header.
Review comment:
done
##########
File path:
integration/spark/src/main/common2.3and2.4/org/apache/spark/sql/CarbonDataSourceScanHelper.scala
##########
@@ -0,0 +1,85 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql
+
+import org.apache.spark.CarbonInputMetrics
+import org.apache.spark.rdd.RDD
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions.{Attribute, Expression =>
SparkExpression}
+import org.apache.spark.sql.catalyst.plans.QueryPlan
+import org.apache.spark.sql.execution.{ColumnarBatchScan, DataSourceScanExec}
+import org.apache.spark.sql.execution.strategy.CarbonPlanHelper
+import org.apache.spark.sql.optimizer.CarbonFilters
+
+import org.apache.carbondata.core.index.IndexFilter
+import org.apache.carbondata.core.indexstore.PartitionSpec
+import org.apache.carbondata.core.scan.expression.Expression
+import org.apache.carbondata.core.scan.expression.logical.AndExpression
+import org.apache.carbondata.hadoop.CarbonProjection
+import org.apache.carbondata.spark.rdd.CarbonScanRDD
+
+abstract class CarbonDataSourceScanHelper(relation:
CarbonDatasourceHadoopRelation,
+ output: Seq[Attribute],
+ partitionFilters: Seq[SparkExpression],
+ pushedDownFilters: Seq[Expression],
+ pushedDownProjection: CarbonProjection,
+ directScanSupport: Boolean,
+ extraRDD: Option[(RDD[InternalRow], Boolean)],
+ segmentIds: Option[String])
+ extends DataSourceScanExec with ColumnarBatchScan {
Review comment:
done
##########
File path:
integration/spark/src/test/scala/org/apache/carbondata/integration/spark/testsuite/binary/TestBinaryDataType.scala
##########
@@ -1551,26 +1552,28 @@ class TestBinaryDataType extends QueryTest with
BeforeAndAfterAll {
sql("insert into hive_table select 'a','b','1'");
}
- val e2 = intercept[SparkException] {
- sql("insert into hive_table2 select 'a','b','binary'");
- }
+ if (!SparkUtil.isSparkVersionXAndAbove("3")) {
+ val e2 = intercept[SparkException] {
+ sql("insert into hive_table2 select 'a','b','binary'");
+ }
- assert(e2.getMessage.contains(
- "Dynamic partition strict mode requires at least one static
partition column"))
+ assert(e2.getMessage.contains(
+ "Dynamic partition strict mode requires at least one static
partition column"))
- val eInt2 = intercept[Exception] {
- sql("insert into hive_table2 select 'a','b','1'");
- }
+ val eInt2 = intercept[Exception] {
+ sql("insert into hive_table2 select 'a','b','1'");
+ }
- val e3 = intercept[SparkException] {
- sql("insert into parquet_table select 'a','b','binary'");
- }
+ val e3 = intercept[SparkException] {
+ sql("insert into parquet_table select 'a','b','binary'");
+ }
- assert(e3.getMessage.contains(
- "Dynamic partition strict mode requires at least one static
partition column"))
+ assert(e3.getMessage.contains(
+ "Dynamic partition strict mode requires at least one static
partition column"))
- val eInt3 = intercept[Exception] {
- sql("insert into parquet_table select 'a','b','1'");
+ val eInt3 = intercept[Exception] {
+ sql("insert into parquet_table select 'a','b','1'");
+ }
Review comment:
it is needed or else spark will not be able to find the path to the
carbon file
##########
File path:
integration/spark/src/test/scala/org/apache/carbondata/spark/testsuite/merge/MergeTestCase.scala
##########
@@ -842,7 +842,7 @@ class MergeTestCase extends QueryTest with
BeforeAndAfterAll {
Row("d", "3")
).asJava, StructType(Seq(StructField("key", StringType),
StructField("value", StringType))))
- initframe.write
+ initframe.repartition(1).write
Review comment:
it is needed or else spark will not be able to find the path to the
carbon file
##########
File path:
mv/plan/src/main/scala/org/apache/carbondata/mv/plans/util/Printers.scala
##########
@@ -358,6 +359,12 @@ trait Printers {
""
}
qualifierPrefix + quoteIdentifier(child.name) + " AS " +
quoteIdentifier(a.name)
+ case a@Alias(child: AggregateExpression, _) =>
+ child.sql + " AS " + quoteIdentifier(a.name)
Review comment:
done
##########
File path:
integration/spark/src/test/scala/org/apache/carbondata/integration/spark/testsuite/binary/TestBinaryDataType.scala
##########
@@ -1551,26 +1552,28 @@ class TestBinaryDataType extends QueryTest with
BeforeAndAfterAll {
sql("insert into hive_table select 'a','b','1'");
}
- val e2 = intercept[SparkException] {
- sql("insert into hive_table2 select 'a','b','binary'");
- }
+ if (!SparkUtil.isSparkVersionXAndAbove("3")) {
+ val e2 = intercept[SparkException] {
+ sql("insert into hive_table2 select 'a','b','binary'");
+ }
- assert(e2.getMessage.contains(
- "Dynamic partition strict mode requires at least one static
partition column"))
+ assert(e2.getMessage.contains(
+ "Dynamic partition strict mode requires at least one static
partition column"))
- val eInt2 = intercept[Exception] {
- sql("insert into hive_table2 select 'a','b','1'");
- }
+ val eInt2 = intercept[Exception] {
+ sql("insert into hive_table2 select 'a','b','1'");
+ }
- val e3 = intercept[SparkException] {
- sql("insert into parquet_table select 'a','b','binary'");
- }
+ val e3 = intercept[SparkException] {
+ sql("insert into parquet_table select 'a','b','binary'");
+ }
- assert(e3.getMessage.contains(
- "Dynamic partition strict mode requires at least one static
partition column"))
+ assert(e3.getMessage.contains(
+ "Dynamic partition strict mode requires at least one static
partition column"))
- val eInt3 = intercept[Exception] {
- sql("insert into parquet_table select 'a','b','1'");
+ val eInt3 = intercept[Exception] {
+ sql("insert into parquet_table select 'a','b','1'");
+ }
Review comment:
if condition has been added, needed for that
##########
File path:
integration/spark/src/main/scala/org/apache/spark/sql/execution/command/management/CarbonShowSegmentsAsSelectCommand.scala
##########
@@ -46,12 +49,25 @@ case class CarbonShowSegmentsAsSelectCommand(
Checker.validateTableExists(databaseNameOp, tableName, sparkSession)
CarbonEnv.getCarbonTable(databaseNameOp, tableName)(sparkSession)
}
- private lazy val df = createDataFrame
+
+ val tempViewName: String = makeTempViewName(carbonTable)
+
+ val df: DataFrame = createDataFrame
override def output: Seq[Attribute] = {
- df.queryExecution.analyzed.output.map { attr =>
- AttributeReference(attr.name, attr.dataType, nullable = true)()
+ var attrList: Seq[Attribute] = Seq()
Review comment:
done
##########
File path:
examples/spark/src/test/scala/org/apache/carbondata/examplesCI/RunExamples.scala
##########
@@ -104,7 +104,8 @@ class RunExamples extends QueryTest with BeforeAndAfterAll {
TableLevelCompactionOptionExample.exampleBody(spark)
}
- test("LuceneIndexExample") {
+ // Below test case ignored due to the Deadlock in spark code
Review comment:
Not reported as it is very specific to Carbondata and this issue is not
reproduced in native spark code without carbondata.
##########
File path:
integration/spark/src/main/scala/org/apache/spark/sql/execution/command/mutation/IUDCommonUtil.scala
##########
@@ -44,7 +44,7 @@ object IUDCommonUtil {
case unresolvedRelation: UnresolvedRelation =>
val dbAndTb =
sparkSession.sessionState.catalog.asInstanceOf[HiveSessionCatalog].getCurrentDatabase
+
- "." + unresolvedRelation.tableIdentifier.table
+ "." +
CarbonToSparkAdapter.getTableIdentifier(unresolvedRelation).get.table
Review comment:
Done
##########
File path:
integration/spark/src/main/scala/org/apache/spark/sql/hive/CarbonAnalysisRules.scala
##########
@@ -135,34 +137,34 @@ case class CarbonIUDAnalysisRule(sparkSession:
SparkSession) extends Rule[Logica
var transformed: Boolean = false
// Create a dummy projection to include filter conditions
var newPlan: LogicalPlan = null
- if (table.tableIdentifier.database.isDefined) {
+ if (tableIdentifier.database.isDefined) {
newPlan = parser.parsePlan("select * from " +
- table.tableIdentifier.database.getOrElse("") + "." +
- table.tableIdentifier.table + " " + alias.getOrElse("") + " " +
filter)
+ tableIdentifier.database.getOrElse("") + "." +
Review comment:
Done
##########
File path:
integration/spark/src/main/scala/org/apache/spark/sql/parser/CarbonSpark2SqlParser.scala
##########
@@ -772,58 +774,62 @@ class CarbonSpark2SqlParser extends CarbonDDLSqlParser {
}
def getFields(schema: Seq[StructField], isExternal: Boolean = false):
Seq[Field] = {
- def getScannerInput(col: StructField,
- columnComment: String,
- columnName: String) = {
- if (col.dataType.catalogString == "float" && !isExternal) {
- '`' + columnName + '`' + " double" + columnComment
+ schema.map { col =>
+ // TODO: Spark has started supporting CharType/VarChar types in Spark
3.1 but both are
+ // marked as experimental. Adding a hack to change to string for now.
Review comment:
Done
##########
File path:
integration/spark/src/main/scala/org/apache/spark/sql/parser/CarbonSpark2SqlParser.scala
##########
@@ -835,10 +841,21 @@ class CarbonSpark2SqlParser extends CarbonDDLSqlParser {
field.dataType = Some("double")
}
field.rawSchema = scannerInput
- if (col.getComment().isDefined) {
+ if (comment.isDefined) {
field.columnComment = plainComment
}
field
+ }
+
+ def getScannerInput(dataType: DataType,
+ columnComment: String,
+ columnName: String,
+ isExternal: Boolean): String = {
Review comment:
done
##########
File path:
integration/spark/src/main/spark3.1/org/apache/spark/sql/CarbonToSparkAdapter.scala
##########
@@ -0,0 +1,263 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql
+
+import java.net.URI
+
+import scala.collection.mutable.ArrayBuffer
+
+import org.apache.spark.{SparkContext, TaskContext}
+import org.apache.spark.scheduler.{SparkListener, SparkListenerApplicationEnd}
+import
org.apache.spark.sql.carbondata.execution.datasources.CarbonFileIndexReplaceRule
+import org.apache.spark.sql.catalyst.catalog.{CatalogStorageFormat,
ExternalCatalogWithListener}
+import org.apache.spark.sql.catalyst.expressions.{Alias, Attribute,
AttributeReference, AttributeSet, Expression, ExprId, NamedExpression,
ScalaUDF, SortOrder, SubqueryExpression}
+import org.apache.spark.sql.catalyst.expressions.codegen._
+import org.apache.spark.sql.catalyst.expressions.codegen.Block._
+import org.apache.spark.sql.catalyst.optimizer.Optimizer
+import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, SubqueryAlias}
+import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.execution.datasources.{DataSourceStrategy,
FilePartition, PartitionedFile}
+import org.apache.spark.sql.hive.HiveExternalCatalog
+import org.apache.spark.sql.optimizer.{CarbonIUDRule, CarbonUDFTransformRule,
MVRewriteRule}
+import org.apache.spark.sql.secondaryindex.optimizer.CarbonSITransformationRule
+import org.apache.spark.sql.types.{DataType, Metadata, StringType}
+
+import org.apache.carbondata.core.util.{CarbonProperties,
ThreadLocalSessionInfo}
+import org.apache.carbondata.geo.{InPolygonJoinUDF, ToRangeListAsStringUDF}
+
+object CarbonToSparkAdapter extends SparkVersionAdapter {
+
+ def createFilePartition(index: Int, files: ArrayBuffer[PartitionedFile]):
FilePartition = {
+ FilePartition(index, files.toArray)
+ }
+
+ def addSparkSessionListener(sparkSession: SparkSession): Unit = {
+ sparkSession.sparkContext.addSparkListener(new SparkListener {
+ override def onApplicationEnd(applicationEnd:
SparkListenerApplicationEnd): Unit = {
+ CarbonEnv.carbonEnvMap.remove(sparkSession)
+ ThreadLocalSessionInfo.unsetAll()
+ }
+ })
+ }
+
+ def addSparkListener(sparkContext: SparkContext): Unit = {
+ sparkContext.addSparkListener(new SparkListener {
+ override def onApplicationEnd(applicationEnd:
SparkListenerApplicationEnd): Unit = {
+ SparkSession.setDefaultSession(null)
+ }
+ })
+ }
+
+ def createAttributeReference(
+ name: String,
+ dataType: DataType,
+ nullable: Boolean,
+ metadata: Metadata,
+ exprId: ExprId,
+ qualifier: Option[String],
+ attrRef : NamedExpression = null): AttributeReference = {
+ val qf = if (qualifier.nonEmpty) Seq(qualifier.get) else Seq.empty
+ AttributeReference(
+ name,
+ dataType,
+ nullable,
+ metadata)(exprId, qf)
+ }
+
+ def createAttributeReference(
+ name: String,
+ dataType: DataType,
+ nullable: Boolean,
+ metadata: Metadata,
+ exprId: ExprId,
+ qualifier: Seq[String]): AttributeReference = {
+ AttributeReference(
+ name,
+ dataType,
+ nullable,
+ metadata)(exprId, qualifier)
+ }
+
+ def lowerCaseAttribute(expression: Expression): Expression =
expression.transform {
+ case attr: AttributeReference =>
+ CarbonToSparkAdapter.createAttributeReference(
+ attr.name.toLowerCase,
+ attr.dataType,
+ attr.nullable,
+ attr.metadata,
+ attr.exprId,
+ attr.qualifier)
+ }
+
+ def createAttributeReference(attr: AttributeReference,
+ attrName: String,
+ newSubsume: String): AttributeReference = {
+ AttributeReference(attrName, attr.dataType)(
+ exprId = attr.exprId,
+ qualifier = newSubsume.split("\n").map(_.trim))
+ }
+
+ def createScalaUDF(s: ScalaUDF, reference: AttributeReference): ScalaUDF = {
+ s.copy(children = Seq(reference))
+ }
+
+ def createExprCode(code: String, isNull: String, value: String, dataType:
DataType): ExprCode = {
+ ExprCode(
+ code"$code",
+ JavaCode.isNullVariable(isNull),
+ JavaCode.variable(value, dataType))
+ }
+
+ def createAliasRef(
+ child: Expression,
+ name: String,
+ exprId: ExprId = NamedExpression.newExprId,
+ qualifier: Seq[String] = Seq.empty,
+ explicitMetadata: Option[Metadata] = None) : Alias = {
+ Alias(child, name)(exprId, qualifier, explicitMetadata)
+ }
+
+ def createAliasRef(
+ child: Expression,
+ name: String,
+ exprId: ExprId,
+ qualifier: Option[String]) : Alias = {
+ Alias(child, name)(exprId,
+ if (qualifier.isEmpty) Seq.empty else Seq(qualifier.get),
+ None)
Review comment:
done
##########
File path:
integration/spark/src/main/spark3.1/org/apache/spark/sql/parser/CarbonSparkSqlParser.scala
##########
@@ -0,0 +1,153 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.spark.sql.parser
+
+import scala.collection.mutable
+
+import org.antlr.v4.runtime.tree.TerminalNode
+import org.apache.spark.sql.{CarbonThreadUtil, CarbonToSparkAdapter,
SparkSession}
+import org.apache.spark.sql.catalyst.parser.{AbstractSqlParser, SqlBaseParser}
+import org.apache.spark.sql.catalyst.parser.SqlBaseParser._
+import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
+import org.apache.spark.sql.execution.{SparkSqlAstBuilder, SparkSqlParser}
+import org.apache.spark.sql.internal.{SQLConf, VariableSubstitution}
+import org.apache.spark.sql.types.StructField
+import org.apache.spark.sql.util.CarbonException
+import org.apache.spark.util.CarbonReflectionUtils
+
+import
org.apache.carbondata.common.exceptions.sql.MalformedCarbonCommandException
+import org.apache.carbondata.spark.util.CarbonScalaUtil
+
+/**
+ * Concrete parser for Spark SQL statements and carbon specific
+ * statements
+ */
+class CarbonSparkSqlParser(conf: SQLConf, sparkSession: SparkSession) extends
SparkSqlParser {
+
+ val parser = new CarbonSpark2SqlParser
+
+ override val astBuilder = CarbonReflectionUtils.getAstBuilder(conf, parser,
sparkSession)
+
+ private val substitutor = new VariableSubstitution
+
+ override def parsePlan(sqlText: String): LogicalPlan = {
+ CarbonThreadUtil.updateSessionInfoToCurrentThread(sparkSession)
+ try {
+ val parsedPlan = super.parsePlan(sqlText)
+ CarbonScalaUtil.cleanParserThreadLocals
+ parsedPlan
+ } catch {
+ case ce: MalformedCarbonCommandException =>
+ CarbonScalaUtil.cleanParserThreadLocals
+ throw ce
+ case ex: Throwable =>
+ try {
+ parser.parse(sqlText)
+ } catch {
+ case mce: MalformedCarbonCommandException =>
+ throw mce
+ case e: Throwable =>
+ CarbonException.analysisException(
+ s"""== Parse1 ==
+ |${ex.getMessage}
+ |== Parse2 ==
+ |${e.getMessage}
+ """.stripMargin.trim)
+ }
+ }
+ }
+
+ protected override def parse[T](command: String)(toResult: SqlBaseParser =>
T): T = {
+ super.parse(substitutor.substitute(command))(toResult)
+ }
+}
+
+class CarbonHelperSqlAstBuilder(conf: SQLConf,
+ parser: CarbonSpark2SqlParser,
+ sparkSession: SparkSession)
+ extends SparkSqlAstBuilderWrapper(conf) {
+ /**
+ * Parse a key-value map from a [[TablePropertyListContext]], assuming all
values are specified.
+ */
+ override def visitPropertyKeyValues(ctx: TablePropertyListContext):
Map[String, String] = {
+ val props = visitTablePropertyList(ctx)
+ CarbonSparkSqlParserUtil.visitPropertyKeyValues(ctx, props)
+ }
+
+ def getPropertyKeyValues(ctx: TablePropertyListContext): Map[String, String]
+ = {
Review comment:
done
##########
File path:
mv/plan/src/main/spark2.3/org/apache/carbondata/mv/plans/modular/ExpressionHelper.scala
##########
@@ -17,7 +17,7 @@
package org.apache.carbondata.mv.plans.modular
-import org.apache.spark.sql.catalyst.expressions.{Alias, AttributeReference,
ExprId, Expression, NamedExpression}
+import org.apache.spark.sql.catalyst.expressions.{Alias, AttributeReference,
Expression, ExprId, NamedExpression}
Review comment:
Checkstyle was not running on this class initially, if changed back then
checkstyle will fail.
--
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.
For queries about this service, please contact Infrastructure at:
[email protected]
