2826

Build # 2826

Build Type

push

jenkins

Committed by GitHub

Commit Message

<a href="https://github.com/apache/carbondata/commit/<a class=hub.com/apache/carbondata/commit/<a class="double-link" href="https://git"><a class=hub.com/apache/carbondata/commit/<a class="double-link" href="https://git"><a class=hub.com/apache/carbondata/commit/<a class="double-link" href="https://git"><a class=hub.com/apache/carbondata/commit/c30cdf46c9fe5cb22e49520bedd75cf98717ff06">c30cdf46c">&lt;a href=&quot;https://github.com/apache/carbondata/commit/</a><a class="double-link" href="https://github.com/apache/carbondata/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/apache/carbondata/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/apache/carbondata/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/apache/carbondata/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/apache/carbondata/commit/c30cdf46c9fe5cb22e49520bedd75cf98717ff06">c30cdf46c</a><a href="https://github.com/apache/carbondata/commit/c30cdf46c9fe5cb22e49520bedd75cf98717ff06">&quot;&gt;&amp;lt;a href=&amp;quot;https://github.com/apache/carbondata/commit/&lt;/a&gt;&lt;a class=&quot;double-link&quot; href=&quot;https://github.com/apache/carbondata/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/apache/carbondata/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/apache/carbondata/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/apache/carbondata/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/apache/carbondata/commit/c30cdf46c9fe5cb22e49520bedd75cf98717ff06&quot;&gt;c30cdf46c&lt;/a&gt;&lt;a href=&quot;https://github.com/apache/carbondata/commit/c30cdf46c9fe5cb22e49520bedd75cf98717ff06&quot;&gt;&amp;quot;&amp;gt;&amp;amp;lt;a href=&amp;amp;quot;https://github.com/apache/carbondata/commit/&amp;lt;/a&amp;gt;&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://github.com/apache/carbondata/commit/&amp;amp;lt;a ... (continued)

Run Details

3 of 3 new or added lines in 1 file covered. (100.0%)

62469 of 78899 relevant lines covered (79.18%)

1.05 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

48.3

/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/stream/CarbonCreateStreamCommand.scala

/*
 * 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.execution.command.stream

import java.util

import scala.collection.JavaConverters._
import scala.collection.mutable

import org.apache.spark.sql._
import org.apache.spark.sql.catalyst.expressions.{Attribute, AttributeReference, Expression, ExprId, NamedExpression}
import org.apache.spark.sql.catalyst.plans.logical.{Filter, LogicalPlan, Project}
import org.apache.spark.sql.execution.command.DataCommand
import org.apache.spark.sql.execution.datasources.LogicalRelation
import org.apache.spark.sql.execution.streaming.StreamingRelation
import org.apache.spark.sql.types.{StringType, StructType}

import org.apache.carbondata.common.exceptions.sql.MalformedCarbonCommandException
import org.apache.carbondata.core.metadata.schema.table.CarbonTable
import org.apache.carbondata.processing.loading.csvinput.CSVInputFormat
import org.apache.carbondata.spark.StreamingOption
import org.apache.carbondata.spark.util.{CarbonSparkUtil, Util}
import org.apache.carbondata.stream.StreamJobManager

/**
 * This command will start a Spark streaming job to insert rows from source to sink
 */
case class CarbonCreateStreamCommand(
    streamName: String,
    sinkDbName: Option[String],
    sinkTableName: String,
    ifNotExists: Boolean,
    optionMap: Map[String, String],
    query: String
) extends DataCommand {

  override def output: Seq[Attribute] =
    Seq(AttributeReference("Stream Name", StringType, nullable = false)(),
      AttributeReference("JobId", StringType, nullable = false)(),
      AttributeReference("Status", StringType, nullable = false)())

  override def processData(sparkSession: SparkSession): Seq[Row] = {
    setAuditTable(CarbonEnv.getDatabaseName(sinkDbName)(sparkSession), sinkTableName)
    setAuditInfo(Map("streamName" -> streamName, "query" -> query) ++ optionMap)
    val inputQuery = sparkSession.sql(query)
    val sourceTableSeq = inputQuery.logicalPlan collect {
      case r: LogicalRelation
        if r.relation.isInstanceOf[CarbonDatasourceHadoopRelation] &&
           r.relation.asInstanceOf[CarbonDatasourceHadoopRelation].carbonTable.isStreamingSource =>
        r.relation.asInstanceOf[CarbonDatasourceHadoopRelation].carbonTable
    }
    if (sourceTableSeq.isEmpty) {
      throw new MalformedCarbonCommandException(
        "Must specify stream source table in the stream query")
    }
    if (sourceTableSeq.size > 1) {
      throw new MalformedCarbonCommandException(
        "Stream query on more than one stream source table is not supported")
    }
    val sourceTable = sourceTableSeq.head

    val tblProperty = sourceTable.getTableInfo.getFactTable.getTableProperties
    val format = sourceTable.getFormat
    if (format == null) {
      throw new MalformedCarbonCommandException("Streaming from carbon file is not supported")
    }
    val updatedQuery = if (format.equals("kafka")) {
      shouldHaveProperty(tblProperty, "kafka.bootstrap.servers", sourceTable)
      shouldHaveProperty(tblProperty, "subscribe", sourceTable)
      createPlan(sparkSession, inputQuery, sourceTable, "kafka", tblProperty.asScala)
    } else if (format.equals("socket")) {
      shouldHaveProperty(tblProperty, "host", sourceTable)
      shouldHaveProperty(tblProperty, "port", sourceTable)
      createPlan(sparkSession, inputQuery, sourceTable, "socket", tblProperty.asScala)
    } else {
      // Replace the logical relation with a streaming relation created
      // from the stream source table
      inputQuery.logicalPlan transform {
        case r: LogicalRelation
          if r.relation.isInstanceOf[CarbonDatasourceHadoopRelation] &&
             r.relation.asInstanceOf[CarbonDatasourceHadoopRelation].carbonTable.isStreamingSource
        => prepareStreamingRelation(sparkSession, r)
        case plan: LogicalPlan => plan
      }
    }

    if (sourceTable == null) {
      throw new MalformedCarbonCommandException("Must specify stream source table in the query")
    }

    // add CSV row parser if user does not specify
    val newMap = mutable.Map[String, String]()
    optionMap.foreach(x => newMap(x._1) = x._2)
    newMap(CSVInputFormat.DELIMITER) = tblProperty.asScala.getOrElse("delimiter", ",")

    // start the streaming job
    val jobId = StreamJobManager.startStream(
      sparkSession = sparkSession,
      ifNotExists = ifNotExists,
      streamName = streamName,
      sourceTable = sourceTable,
      sinkTable = CarbonEnv.getCarbonTable(sinkDbName, sinkTableName)(sparkSession),
      query = query,
      streamDf = Dataset.ofRows(sparkSession, updatedQuery),
      options = new StreamingOption(newMap.toMap)
    )
    Seq(Row(streamName, jobId, "RUNNING"))
  }

  /**
   * Create a new plan for the stream query on kafka and Socket source table.
   * This is required because we need to convert the schema of the data stored in kafka
   * The returned logical plan contains the complete plan tree of original plan with
   * logical relation replaced with a streaming relation.
   *
   * @param sparkSession spark session
   * @param inputQuery stream query from user
   * @param sourceTable source table (kafka table)
   * @param sourceName source name, kafka or socket
   * @param tblProperty table property of source table
   * @return a new logical plan
   */
  private def createPlan(
      sparkSession: SparkSession,
      inputQuery: DataFrame,
      sourceTable: CarbonTable,
      sourceName: String,
      tblProperty: mutable.Map[String, String]): LogicalPlan = {
    // We follow 3 steps to generate new plan
    // 1. replace the logical relation in stream query with streaming relation
    // 2. collect the new ExprId generated
    // 3. update the stream query plan with the new ExprId generated, to make the plan consistent

    // exprList is used for UDF to extract the data from the 'value' column in kafka
    val columnNames = Util.convertToSparkSchema(sourceTable).fieldNames
    val exprList = columnNames.zipWithIndex.map {
      case (columnName, i) =>
        s"case when size(_values) > $i then _values[$i] else null end AS $columnName"
    }

    val aliasMap = new util.HashMap[String, ExprId]()
    val updatedQuery = inputQuery.logicalPlan transform {
      case r: LogicalRelation
        if r.relation.isInstanceOf[CarbonDatasourceHadoopRelation] &&
           r.relation.asInstanceOf[CarbonDatasourceHadoopRelation].carbonTable.isStreamingSource =>
        // for kafka stream source, get the 'value' column and split it by using UDF
        val plan = sparkSession.readStream
          .format(sourceName)
          .options(tblProperty)
          .load()
          .selectExpr("CAST(value as string) as _value")
        val recordFormat = tblProperty.getOrElse("record_format", "csv")
        val newPlan = recordFormat match {
            case "csv" =>
              val delimiter = tblProperty.getOrElse("delimiter", ",")
              plan.selectExpr(
                s"split(_value, '${CarbonSparkUtil.delimiterConverter4Udf(delimiter)}') as _values")
                .selectExpr(exprList: _*)
                .logicalPlan
            case "json" =>
              import org.apache.spark.sql.functions._
              plan
                .select(from_json(col("_value"), Util.convertToSparkSchema(sourceTable)) as "_data")
                .select("_data.*")
                .logicalPlan
          }

        // collect the newly generated ExprId
        newPlan collect {
          case p@Project(projectList, child) =>
            projectList.map { expr =>
              aliasMap.put(expr.name, expr.exprId)
            }
            p
        }
        newPlan
      case plan: LogicalPlan => plan
    }

    // transform the stream plan to replace all attribute with the collected ExprId
    val transFormedPlan = updatedQuery transform {
      case p@Project(projectList: Seq[NamedExpression], child) =>
        val newProjectList = projectList.map { expr =>
          val newExpr = expr transform {
            case attribute: Attribute =>
              val exprId: ExprId = aliasMap.get(attribute.name)
              if (exprId != null) {
                if (exprId.id != attribute.exprId.id) {
                  AttributeReference(
                    attribute.name, attribute.dataType, attribute.nullable,
                    attribute.metadata)(exprId, attribute.qualifier)
                } else {
                  attribute
                }
              } else {
                attribute
              }
          }
          newExpr.asInstanceOf[NamedExpression]
        }
        Project(newProjectList, child)
      case f@Filter(condition: Expression, child) =>
        val newCondition = condition transform {
          case attribute: Attribute =>
            val exprId: ExprId = aliasMap.get(attribute.name)
            if (exprId != null) {
              if (exprId.id != attribute.exprId.id) {
                AttributeReference(
                  attribute.name, attribute.dataType, attribute.nullable,
                  attribute.metadata)(exprId, attribute.qualifier)
              } else {
                attribute
              }
            } else {
              attribute
            }
        }
        Filter(newCondition, child)
    }
    transFormedPlan
  }

  /**
   * Create a streaming relation from the input logical relation (source table)
   *
   * @param sparkSession spark session
   * @param logicalRelation source table to convert
   * @return sourceTable and its streaming relation
   */
  private def prepareStreamingRelation(
      sparkSession: SparkSession,
      logicalRelation: LogicalRelation): StreamingRelation = {
    val sourceTable = logicalRelation.relation
      .asInstanceOf[CarbonDatasourceHadoopRelation].carbonTable
    val tblProperty = sourceTable.getTableInfo.getFactTable.getTableProperties
    val format = sourceTable.getFormat
    if (format == null) {
      throw new MalformedCarbonCommandException("Streaming from carbon file is not supported")
    }
    val streamReader = sparkSession
      .readStream
      .schema(getSparkSchema(sourceTable))
      .format(format)
    val dataFrame: DataFrame = format match {
      case "csv" | "text" | "json" | "parquet" =>
        shouldHaveProperty(tblProperty, "path", sourceTable)
        streamReader.load(tblProperty.get("path"))
      case other =>
        throw new MalformedCarbonCommandException(s"Streaming from $format is not supported")
    }
    val streamRelation = dataFrame.logicalPlan.asInstanceOf[StreamingRelation]

    // Since SparkSQL analyzer will match the UUID in attribute,
    // create a new StreamRelation and re-use the same attribute from LogicalRelation
    StreamingRelation(streamRelation.dataSource, streamRelation.sourceName, logicalRelation.output)
  }

  private def shouldHaveProperty(
      tblProperty: java.util.Map[String, String],
      propertyName: String,
      sourceTable: CarbonTable) : Unit = {
    if (!tblProperty.containsKey(propertyName)) {
      throw new MalformedCarbonCommandException(
        s"tblproperty '$propertyName' should be provided for stream source " +
        s"${sourceTable.getDatabaseName}.${sourceTable.getTableName}")
    }
  }

  private def getSparkSchema(sourceTable: CarbonTable): StructType = {
    val cols = sourceTable.getTableInfo.getFactTable.getListOfColumns.asScala.toArray
    val sortedCols = cols.filter(_.getSchemaOrdinal != -1)
      .sortWith(_.getSchemaOrdinal < _.getSchemaOrdinal)
    Util.convertToSparkSchema(sourceTable, sortedCols)
  }

  override protected def opName: String = "CREATE STREAM"
}

1	/*
2	* Licensed to the Apache Software Foundation (ASF) under one or more
3	* contributor license agreements. See the NOTICE file distributed with
4	* this work for additional information regarding copyright ownership.
5	* The ASF licenses this file to You under the Apache License, Version 2.0
6	* (the "License"); you may not use this file except in compliance with
7	* the License. You may obtain a copy of the License at
8	*
9	* http://www.apache.org/licenses/LICENSE-2.0
10	*
11	* Unless required by applicable law or agreed to in writing, software
12	* distributed under the License is distributed on an "AS IS" BASIS,
13	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14	* See the License for the specific language governing permissions and
15	* limitations under the License.
16	*/
17
18	package org.apache.spark.sql.execution.command.stream
19
20	import java.util
21
22	import scala.collection.JavaConverters._
23	import scala.collection.mutable
24
25	import org.apache.spark.sql._
26	import org.apache.spark.sql.catalyst.expressions.{Attribute, AttributeReference, Expression, ExprId, NamedExpression}
27	import org.apache.spark.sql.catalyst.plans.logical.{Filter, LogicalPlan, Project}
28	import org.apache.spark.sql.execution.command.DataCommand
29	import org.apache.spark.sql.execution.datasources.LogicalRelation
30	import org.apache.spark.sql.execution.streaming.StreamingRelation
31	import org.apache.spark.sql.types.{StringType, StructType}
32
33	import org.apache.carbondata.common.exceptions.sql.MalformedCarbonCommandException
34	import org.apache.carbondata.core.metadata.schema.table.CarbonTable
35	import org.apache.carbondata.processing.loading.csvinput.CSVInputFormat
36	import org.apache.carbondata.spark.StreamingOption
37	import org.apache.carbondata.spark.util.{CarbonSparkUtil, Util}
38	import org.apache.carbondata.stream.StreamJobManager
39
40	/**
41	* This command will start a Spark streaming job to insert rows from source to sink
42	*/
43	case class CarbonCreateStreamCommand(	2✔
44	streamName: String,	2✔
45	sinkDbName: Option[String],	2✔
46	sinkTableName: String,	2✔
47	ifNotExists: Boolean,	2✔
48	optionMap: Map[String, String],	2✔
49	query: String	2✔
50	) extends DataCommand {	2✔
51
52	override def output: Seq[Attribute] =
53	Seq(AttributeReference("Stream Name", StringType, nullable = false)(),	2✔
54	AttributeReference("JobId", StringType, nullable = false)(),	2✔
55	AttributeReference("Status", StringType, nullable = false)())	2✔
56
57	override def processData(sparkSession: SparkSession): Seq[Row] = {
58	setAuditTable(CarbonEnv.getDatabaseName(sinkDbName)(sparkSession), sinkTableName)	2✔
59	setAuditInfo(Map("streamName" -> streamName, "query" -> query) ++ optionMap)	2✔
60	val inputQuery = sparkSession.sql(query)	2✔
61	val sourceTableSeq = inputQuery.logicalPlan collect {	2✔
62	case r: LogicalRelation	2✔
63	if r.relation.isInstanceOf[CarbonDatasourceHadoopRelation] &&	2✔
64	r.relation.asInstanceOf[CarbonDatasourceHadoopRelation].carbonTable.isStreamingSource =>	2✔
65	r.relation.asInstanceOf[CarbonDatasourceHadoopRelation].carbonTable	2✔
66	}
67	if (sourceTableSeq.isEmpty) {	2✔
68	throw new MalformedCarbonCommandException(	2✔
69	"Must specify stream source table in the stream query")	2✔
70	}
71	if (sourceTableSeq.size > 1) {	2✔
72	throw new MalformedCarbonCommandException(	×
73	"Stream query on more than one stream source table is not supported")	×
74	}
75	val sourceTable = sourceTableSeq.head	2✔
76
77	val tblProperty = sourceTable.getTableInfo.getFactTable.getTableProperties	2✔
78	val format = sourceTable.getFormat	2✔
79	if (format == null) {	2✔
80	throw new MalformedCarbonCommandException("Streaming from carbon file is not supported")	2✔
81	}
82	val updatedQuery = if (format.equals("kafka")) {	2✔
83	shouldHaveProperty(tblProperty, "kafka.bootstrap.servers", sourceTable)	×
84	shouldHaveProperty(tblProperty, "subscribe", sourceTable)	×
85	createPlan(sparkSession, inputQuery, sourceTable, "kafka", tblProperty.asScala)	×
86	} else if (format.equals("socket")) {	2✔
87	shouldHaveProperty(tblProperty, "host", sourceTable)	×
88	shouldHaveProperty(tblProperty, "port", sourceTable)	×
89	createPlan(sparkSession, inputQuery, sourceTable, "socket", tblProperty.asScala)	×
90	} else {
91	// Replace the logical relation with a streaming relation created
92	// from the stream source table
93	inputQuery.logicalPlan transform {	2✔
94	case r: LogicalRelation	2✔
95	if r.relation.isInstanceOf[CarbonDatasourceHadoopRelation] &&	2✔
96	r.relation.asInstanceOf[CarbonDatasourceHadoopRelation].carbonTable.isStreamingSource	2✔
97	=> prepareStreamingRelation(sparkSession, r)	2✔
98	case plan: LogicalPlan => plan	2✔
99	}
100	}
101
102	if (sourceTable == null) {	2✔
103	throw new MalformedCarbonCommandException("Must specify stream source table in the query")	×
104	}
105
106	// add CSV row parser if user does not specify
107	val newMap = mutable.Map[String, String]()	2✔
108	optionMap.foreach(x => newMap(x._1) = x._2)	2✔
109	newMap(CSVInputFormat.DELIMITER) = tblProperty.asScala.getOrElse("delimiter", ",")	2✔
110
111	// start the streaming job
112	val jobId = StreamJobManager.startStream(	2✔
113	sparkSession = sparkSession,	2✔
114	ifNotExists = ifNotExists,	2✔
115	streamName = streamName,	2✔
116	sourceTable = sourceTable,	2✔
117	sinkTable = CarbonEnv.getCarbonTable(sinkDbName, sinkTableName)(sparkSession),	2✔
118	query = query,	2✔
119	streamDf = Dataset.ofRows(sparkSession, updatedQuery),	2✔
120	options = new StreamingOption(newMap.toMap)	2✔
121	)
122	Seq(Row(streamName, jobId, "RUNNING"))	2✔
123	}
124
125	/**
126	* Create a new plan for the stream query on kafka and Socket source table.
127	* This is required because we need to convert the schema of the data stored in kafka
128	* The returned logical plan contains the complete plan tree of original plan with
129	* logical relation replaced with a streaming relation.
130	*
131	* @param sparkSession spark session
132	* @param inputQuery stream query from user
133	* @param sourceTable source table (kafka table)
134	* @param sourceName source name, kafka or socket
135	* @param tblProperty table property of source table
136	* @return a new logical plan
137	*/
138	private def createPlan(
139	sparkSession: SparkSession,
140	inputQuery: DataFrame,
141	sourceTable: CarbonTable,
142	sourceName: String,
143	tblProperty: mutable.Map[String, String]): LogicalPlan = {
144	// We follow 3 steps to generate new plan
145	// 1. replace the logical relation in stream query with streaming relation
146	// 2. collect the new ExprId generated
147	// 3. update the stream query plan with the new ExprId generated, to make the plan consistent
148
149	// exprList is used for UDF to extract the data from the 'value' column in kafka
150	val columnNames = Util.convertToSparkSchema(sourceTable).fieldNames	×
151	val exprList = columnNames.zipWithIndex.map {	×
152	case (columnName, i) =>	×
153	s"case when size(_values) > $i then _values[$i] else null end AS $columnName"	×
154	}
155
156	val aliasMap = new util.HashMap[String, ExprId]()	×
157	val updatedQuery = inputQuery.logicalPlan transform {	×
158	case r: LogicalRelation	×
159	if r.relation.isInstanceOf[CarbonDatasourceHadoopRelation] &&	×
160	r.relation.asInstanceOf[CarbonDatasourceHadoopRelation].carbonTable.isStreamingSource =>	×
161	// for kafka stream source, get the 'value' column and split it by using UDF
162	val plan = sparkSession.readStream	×
163	.format(sourceName)	×
164	.options(tblProperty)	×
165	.load()	×
166	.selectExpr("CAST(value as string) as _value")	×
167	val recordFormat = tblProperty.getOrElse("record_format", "csv")	×
168	val newPlan = recordFormat match {	×
169	case "csv" =>	×
170	val delimiter = tblProperty.getOrElse("delimiter", ",")	×
171	plan.selectExpr(	×
172	s"split(_value, '${CarbonSparkUtil.delimiterConverter4Udf(delimiter)}') as _values")	×
173	.selectExpr(exprList: _*)	×
174	.logicalPlan	×
175	case "json" =>	×
176	import org.apache.spark.sql.functions._
177	plan	×
178	.select(from_json(col("_value"), Util.convertToSparkSchema(sourceTable)) as "_data")	×
179	.select("_data.*")	×
180	.logicalPlan	×
181	}
182
183	// collect the newly generated ExprId
184	newPlan collect {	×
185	case p@Project(projectList, child) =>	×
186	projectList.map { expr =>	×
187	aliasMap.put(expr.name, expr.exprId)	×
188	}
189	p	×
190	}
191	newPlan	×
192	case plan: LogicalPlan => plan	×
193	}
194
195	// transform the stream plan to replace all attribute with the collected ExprId
196	val transFormedPlan = updatedQuery transform {	×
197	case p@Project(projectList: Seq[NamedExpression], child) =>	×
198	val newProjectList = projectList.map { expr =>	×
199	val newExpr = expr transform {	×
200	case attribute: Attribute =>	×
201	val exprId: ExprId = aliasMap.get(attribute.name)	×
202	if (exprId != null) {	×
203	if (exprId.id != attribute.exprId.id) {	×
204	AttributeReference(	×
205	attribute.name, attribute.dataType, attribute.nullable,	×
206	attribute.metadata)(exprId, attribute.qualifier)	×
207	} else {
208	attribute	×
209	}
210	} else {
211	attribute	×
212	}
213	}
214	newExpr.asInstanceOf[NamedExpression]	×
215	}
216	Project(newProjectList, child)	×
217	case f@Filter(condition: Expression, child) =>	×
218	val newCondition = condition transform {	×
219	case attribute: Attribute =>	×
220	val exprId: ExprId = aliasMap.get(attribute.name)	×
221	if (exprId != null) {	×
222	if (exprId.id != attribute.exprId.id) {	×
223	AttributeReference(	×
224	attribute.name, attribute.dataType, attribute.nullable,	×
225	attribute.metadata)(exprId, attribute.qualifier)	×
226	} else {
227	attribute	×
228	}
229	} else {
230	attribute	×
231	}
232	}
233	Filter(newCondition, child)	×
234	}
235	transFormedPlan	×
236	}
237
238	/**
239	* Create a streaming relation from the input logical relation (source table)
240	*
241	* @param sparkSession spark session
242	* @param logicalRelation source table to convert
243	* @return sourceTable and its streaming relation
244	*/
245	private def prepareStreamingRelation(
246	sparkSession: SparkSession,
247	logicalRelation: LogicalRelation): StreamingRelation = {
248	val sourceTable = logicalRelation.relation	2✔
249	.asInstanceOf[CarbonDatasourceHadoopRelation].carbonTable	2✔
250	val tblProperty = sourceTable.getTableInfo.getFactTable.getTableProperties	2✔
251	val format = sourceTable.getFormat	2✔
252	if (format == null) {	2✔
253	throw new MalformedCarbonCommandException("Streaming from carbon file is not supported")	×
254	}
255	val streamReader = sparkSession	2✔
256	.readStream	2✔
257	.schema(getSparkSchema(sourceTable))	2✔
258	.format(format)	2✔
259	val dataFrame: DataFrame = format match {	2✔
260	case "csv" \| "text" \| "json" \| "parquet" =>	2✔
261	shouldHaveProperty(tblProperty, "path", sourceTable)	2✔
262	streamReader.load(tblProperty.get("path"))	2✔
263	case other =>
264	throw new MalformedCarbonCommandException(s"Streaming from $format is not supported")	×
265	}
266	val streamRelation = dataFrame.logicalPlan.asInstanceOf[StreamingRelation]	2✔
267
268	// Since SparkSQL analyzer will match the UUID in attribute,
269	// create a new StreamRelation and re-use the same attribute from LogicalRelation
270	StreamingRelation(streamRelation.dataSource, streamRelation.sourceName, logicalRelation.output)	2✔
271	}
272
273	private def shouldHaveProperty(
274	tblProperty: java.util.Map[String, String],
275	propertyName: String,
276	sourceTable: CarbonTable) : Unit = {
277	if (!tblProperty.containsKey(propertyName)) {	2✔
278	throw new MalformedCarbonCommandException(	×
279	s"tblproperty '$propertyName' should be provided for stream source " +	×
280	s"${sourceTable.getDatabaseName}.${sourceTable.getTableName}")	×
281	}
282	}
283
284	private def getSparkSchema(sourceTable: CarbonTable): StructType = {
285	val cols = sourceTable.getTableInfo.getFactTable.getListOfColumns.asScala.toArray	2✔
286	val sortedCols = cols.filter(_.getSchemaOrdinal != -1)	2✔
287	.sortWith(_.getSchemaOrdinal < _.getSchemaOrdinal)	2✔
288	Util.convertToSparkSchema(sourceTable, sortedCols)	2✔
289	}
290
291	override protected def opName: String = "CREATE STREAM"	2✔
292	}

apache / carbondata / 2826

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous