2475

Build # 2475

Build Type

push

jenkins

Committed by GitHub

Commit Message

<a href="https://github.com/apache/carbondata/commit/<a class=hub.com/apache/carbondata/commit/a28dacda0a4d8c682aa1b3bae43d5020d8b021db">a28dacda0<a href="https://github.com/apache/carbondata/commit/a28dacda0a4d8c682aa1b3bae43d5020d8b021db">">Merge </a><a class="double-link" href="https://github.com/apache/carbondata/commit/<a class="double-link" href="https://github.com/apache/carbondata/commit/e831623afbc62b96decfb96c9457ab1d5fbdede3">e831623af</a>">e831623af</a><a href="https://github.com/apache/carbondata/commit/a28dacda0a4d8c682aa1b3bae43d5020d8b021db"> into d78db8f6b">d78db8f6b</a>

Run Details

62477 of 78721 relevant lines covered (79.37%)

1.05 hits per line

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

88.97

/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/mutation/CarbonProjectForUpdateCommand.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.mutation

import org.apache.spark.sql._
import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, Project}
import org.apache.spark.sql.execution.command._
import org.apache.spark.sql.execution.command.management.CarbonLoadDataCommand
import org.apache.spark.sql.execution.datasources.LogicalRelation
import org.apache.spark.sql.types.ArrayType
import org.apache.spark.storage.StorageLevel

import org.apache.carbondata.common.exceptions.sql.MalformedCarbonCommandException
import org.apache.carbondata.common.logging.LogServiceFactory
import org.apache.carbondata.core.constants.CarbonCommonConstants
import org.apache.carbondata.core.datamap.Segment
import org.apache.carbondata.core.exception.ConcurrentOperationException
import org.apache.carbondata.core.features.TableOperation
import org.apache.carbondata.core.locks.{CarbonLockFactory, CarbonLockUtil, LockUsage}
import org.apache.carbondata.core.metadata.schema.table.CarbonTable
import org.apache.carbondata.core.mutate.CarbonUpdateUtil
import org.apache.carbondata.core.statusmanager.SegmentStatusManager
import org.apache.carbondata.core.util.CarbonProperties
import org.apache.carbondata.events.{OperationContext, OperationListenerBus, UpdateTablePostEvent, UpdateTablePreEvent}
import org.apache.carbondata.processing.loading.FailureCauses

private[sql] case class CarbonProjectForUpdateCommand(
    plan: LogicalPlan,
    databaseNameOp: Option[String],
    tableName: String,
    columns: List[String])
  extends DataCommand {

  override def processData(sparkSession: SparkSession): Seq[Row] = {
    val LOGGER = LogServiceFactory.getLogService(this.getClass.getName)
    IUDCommonUtil.checkIfSegmentListIsSet(sparkSession, plan)
    val res = plan find {
      case relation: LogicalRelation if relation.relation
        .isInstanceOf[CarbonDatasourceHadoopRelation] =>
        true
      case _ => false
    }

    if (res.isEmpty) {
      return Seq.empty
    }
    val carbonTable = CarbonEnv.getCarbonTable(databaseNameOp, tableName)(sparkSession)
    setAuditTable(carbonTable)
    setAuditInfo(Map("plan" -> plan.simpleString))
    columns.foreach { col =>
      val dataType = carbonTable.getColumnByName(tableName, col).getColumnSchema.getDataType
      if (dataType.isComplexType) {
        throw new UnsupportedOperationException("Unsupported operation on Complex data type")
      }

    }
    if (!carbonTable.getTableInfo.isTransactionalTable) {
      throw new MalformedCarbonCommandException("Unsupported operation on non transactional table")
    }
    if (SegmentStatusManager.isCompactionInProgress(carbonTable)) {
      throw new ConcurrentOperationException(carbonTable, "compaction", "data update")
    }
    if (SegmentStatusManager.isLoadInProgressInTable(carbonTable)) {
      throw new ConcurrentOperationException(carbonTable, "loading", "data update")
    }

    if (!carbonTable.canAllow(carbonTable, TableOperation.UPDATE)) {
      throw new MalformedCarbonCommandException(
        "update operation is not supported for index datamap")
    }

    // trigger event for Update table
    val operationContext = new OperationContext
    val updateTablePreEvent: UpdateTablePreEvent =
      UpdateTablePreEvent(sparkSession, carbonTable)
    OperationListenerBus.getInstance.fireEvent(updateTablePreEvent, operationContext)
    val metadataLock = CarbonLockFactory
      .getCarbonLockObj(carbonTable.getAbsoluteTableIdentifier,
        LockUsage.METADATA_LOCK)
    var lockStatus = false
    // get the current time stamp which should be same for delete and update.
    val currentTime = CarbonUpdateUtil.readCurrentTime
    //    var dataFrame: DataFrame = null
    var dataSet: DataFrame = null
    val isPersistEnabled = CarbonProperties.getInstance.isPersistUpdateDataset
    try {
      lockStatus = metadataLock.lockWithRetries()
      if (lockStatus) {
        logInfo("Successfully able to get the table metadata file lock")
      }
      else {
        throw new Exception("Table is locked for updation. Please try after some time")
      }
      // Get RDD.

      dataSet = if (isPersistEnabled) {
        Dataset.ofRows(sparkSession, plan).persist(StorageLevel.fromString(
          CarbonProperties.getInstance.getUpdateDatasetStorageLevel()))
      }
      else {
        Dataset.ofRows(sparkSession, plan)
      }
      val executionErrors = new ExecutionErrors(FailureCauses.NONE, "")


      // handle the clean up of IUD.
      CarbonUpdateUtil.cleanUpDeltaFiles(carbonTable, false)

      // do delete operation.
      val segmentsToBeDeleted = DeleteExecution.deleteDeltaExecution(
        databaseNameOp,
        tableName,
        sparkSession,
        dataSet.rdd,
        currentTime + "",
        isUpdateOperation = true,
        executionErrors)

      if (executionErrors.failureCauses != FailureCauses.NONE) {
        throw new Exception(executionErrors.errorMsg)
      }

      // do update operation.
      performUpdate(dataSet,
        databaseNameOp,
        tableName,
        plan,
        sparkSession,
        currentTime,
        executionErrors,
        segmentsToBeDeleted)

      if (executionErrors.failureCauses != FailureCauses.NONE) {
        throw new Exception(executionErrors.errorMsg)
      }

      // Do IUD Compaction.
      HorizontalCompaction.tryHorizontalCompaction(
        sparkSession, carbonTable, isUpdateOperation = true)

      // trigger event for Update table
      val updateTablePostEvent: UpdateTablePostEvent =
        UpdateTablePostEvent(sparkSession, carbonTable)
      OperationListenerBus.getInstance.fireEvent(updateTablePostEvent, operationContext)
    } catch {
      case e: HorizontalCompactionException =>
        LOGGER.error(
          "Update operation passed. Exception in Horizontal Compaction. Please check logs." + e)
        // In case of failure , clean all related delta files
        CarbonUpdateUtil.cleanStaleDeltaFiles(carbonTable, e.compactionTimeStamp.toString)

      case e: Exception =>
        LOGGER.error("Exception in update operation", e)
        // ****** start clean up.
        // In case of failure , clean all related delete delta files
        CarbonUpdateUtil.cleanStaleDeltaFiles(carbonTable, currentTime + "")

        // *****end clean up.
        if (null != e.getMessage) {
          sys.error("Update operation failed. " + e.getMessage)
        }
        if (null != e.getCause && null != e.getCause.getMessage) {
          sys.error("Update operation failed. " + e.getCause.getMessage)
        }
        sys.error("Update operation failed. please check logs.")
    }
    finally {
      if (null != dataSet && isPersistEnabled) {
        dataSet.unpersist()
      }
      if (lockStatus) {
        CarbonLockUtil.fileUnlock(metadataLock, LockUsage.METADATA_LOCK)
      }
    }
    Seq.empty
  }

  private def performUpdate(
      dataFrame: Dataset[Row],
      databaseNameOp: Option[String],
      tableName: String,
      plan: LogicalPlan,
      sparkSession: SparkSession,
      currentTime: Long,
      executorErrors: ExecutionErrors,
      deletedSegments: Seq[Segment]): Unit = {

    def isDestinationRelation(relation: CarbonDatasourceHadoopRelation): Boolean = {
      val dbName = CarbonEnv.getDatabaseName(databaseNameOp)(sparkSession)
      (databaseNameOp.isDefined &&
       databaseNameOp.get == dbName &&
       tableName == relation.identifier.getCarbonTableIdentifier.getTableName) ||
      (tableName == relation.identifier.getCarbonTableIdentifier.getTableName)
    }

    // from the dataFrame schema iterate through all the column to be updated and
    // check for the data type, if the data type is complex then throw exception
    def checkForUnsupportedDataType(dataFrame: DataFrame): Unit = {
      dataFrame.schema.foreach(col => {
        // the new column to be updated will be appended with "-updatedColumn" suffix
        if (col.name.endsWith(CarbonCommonConstants.UPDATED_COL_EXTENSION) &&
            col.dataType.isInstanceOf[ArrayType]) {
          throw new UnsupportedOperationException("Unsupported data type: Array")
        }
      })
    }

    def getHeader(relation: CarbonDatasourceHadoopRelation, plan: LogicalPlan): String = {
      var header = ""
      var found = false

      plan match {
        case Project(pList, _) if (!found) =>
          found = true
          header = pList
            .filter(field => !field.name
              .equalsIgnoreCase(CarbonCommonConstants.CARBON_IMPLICIT_COLUMN_TUPLEID))
            .map(col => if (col.name.endsWith(CarbonCommonConstants.UPDATED_COL_EXTENSION)) {
              col.name
                .substring(0, col.name.lastIndexOf(CarbonCommonConstants.UPDATED_COL_EXTENSION))
            }
            else {
              col.name
            }).mkString(",")
      }
      header
    }

    // check for the data type of the new value to be updated
    checkForUnsupportedDataType(dataFrame)
    val ex = dataFrame.queryExecution.analyzed
    val res = ex find {
      case relation: LogicalRelation
        if relation.relation.isInstanceOf[CarbonDatasourceHadoopRelation] &&
           isDestinationRelation(relation.relation.asInstanceOf[CarbonDatasourceHadoopRelation]) =>
        true
      case _ => false
    }
    val carbonRelation: CarbonDatasourceHadoopRelation = res match {
      case Some(relation: LogicalRelation) =>
        relation.relation.asInstanceOf[CarbonDatasourceHadoopRelation]
      case _ => sys.error("")
    }

    val updateTableModel = UpdateTableModel(true, currentTime, executorErrors, deletedSegments)

    val header = getHeader(carbonRelation, plan)

    CarbonLoadDataCommand(
      Some(carbonRelation.identifier.getCarbonTableIdentifier.getDatabaseName),
      carbonRelation.identifier.getCarbonTableIdentifier.getTableName,
      null,
      Seq(),
      Map(("fileheader" -> header)),
      false,
      null,
      Some(dataFrame),
      Some(updateTableModel)).run(sparkSession)

    executorErrors.errorMsg = updateTableModel.executorErrors.errorMsg
    executorErrors.failureCauses = updateTableModel.executorErrors.failureCauses

    Seq.empty

  }

  override protected def opName: String = "UPDATE DATA"
}

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.mutation
19
20	import org.apache.spark.sql._
21	import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, Project}
22	import org.apache.spark.sql.execution.command._
23	import org.apache.spark.sql.execution.command.management.CarbonLoadDataCommand
24	import org.apache.spark.sql.execution.datasources.LogicalRelation
25	import org.apache.spark.sql.types.ArrayType
26	import org.apache.spark.storage.StorageLevel
27
28	import org.apache.carbondata.common.exceptions.sql.MalformedCarbonCommandException
29	import org.apache.carbondata.common.logging.LogServiceFactory
30	import org.apache.carbondata.core.constants.CarbonCommonConstants
31	import org.apache.carbondata.core.datamap.Segment
32	import org.apache.carbondata.core.exception.ConcurrentOperationException
33	import org.apache.carbondata.core.features.TableOperation
34	import org.apache.carbondata.core.locks.{CarbonLockFactory, CarbonLockUtil, LockUsage}
35	import org.apache.carbondata.core.metadata.schema.table.CarbonTable
36	import org.apache.carbondata.core.mutate.CarbonUpdateUtil
37	import org.apache.carbondata.core.statusmanager.SegmentStatusManager
38	import org.apache.carbondata.core.util.CarbonProperties
39	import org.apache.carbondata.events.{OperationContext, OperationListenerBus, UpdateTablePostEvent, UpdateTablePreEvent}
40	import org.apache.carbondata.processing.loading.FailureCauses
41
42	private[sql] case class CarbonProjectForUpdateCommand(	2✔
43	plan: LogicalPlan,	2✔
44	databaseNameOp: Option[String],	2✔
45	tableName: String,	2✔
46	columns: List[String])	2✔
47	extends DataCommand {	2✔
48
49	override def processData(sparkSession: SparkSession): Seq[Row] = {
50	val LOGGER = LogServiceFactory.getLogService(this.getClass.getName)	2✔
51	IUDCommonUtil.checkIfSegmentListIsSet(sparkSession, plan)	2✔
52	val res = plan find {	2✔
53	case relation: LogicalRelation if relation.relation	2✔
54	.isInstanceOf[CarbonDatasourceHadoopRelation] =>
55	true	2✔
56	case _ => false	2✔
57	}
58
59	if (res.isEmpty) {	2✔
60	return Seq.empty	×
61	}
62	val carbonTable = CarbonEnv.getCarbonTable(databaseNameOp, tableName)(sparkSession)	2✔
63	setAuditTable(carbonTable)	2✔
64	setAuditInfo(Map("plan" -> plan.simpleString))	2✔
65	columns.foreach { col =>	2✔
66	val dataType = carbonTable.getColumnByName(tableName, col).getColumnSchema.getDataType	2✔
67	if (dataType.isComplexType) {	2✔
68	throw new UnsupportedOperationException("Unsupported operation on Complex data type")	1✔
69	}
70
71	}
72	if (!carbonTable.getTableInfo.isTransactionalTable) {	2✔
73	throw new MalformedCarbonCommandException("Unsupported operation on non transactional table")	1✔
74	}
75	if (SegmentStatusManager.isCompactionInProgress(carbonTable)) {	2✔
76	throw new ConcurrentOperationException(carbonTable, "compaction", "data update")	×
77	}
78	if (SegmentStatusManager.isLoadInProgressInTable(carbonTable)) {	2✔
79	throw new ConcurrentOperationException(carbonTable, "loading", "data update")	×
80	}
81
82	if (!carbonTable.canAllow(carbonTable, TableOperation.UPDATE)) {	2✔
83	throw new MalformedCarbonCommandException(	×
84	"update operation is not supported for index datamap")	×
85	}
86
87	// trigger event for Update table
88	val operationContext = new OperationContext	2✔
89	val updateTablePreEvent: UpdateTablePreEvent =	2✔
90	UpdateTablePreEvent(sparkSession, carbonTable)	2✔
91	OperationListenerBus.getInstance.fireEvent(updateTablePreEvent, operationContext)	2✔
92	val metadataLock = CarbonLockFactory	2✔
93	.getCarbonLockObj(carbonTable.getAbsoluteTableIdentifier,	2✔
94	LockUsage.METADATA_LOCK)	2✔
95	var lockStatus = false	2✔
96	// get the current time stamp which should be same for delete and update.
97	val currentTime = CarbonUpdateUtil.readCurrentTime	2✔
98	// var dataFrame: DataFrame = null
99	var dataSet: DataFrame = null	2✔
100	val isPersistEnabled = CarbonProperties.getInstance.isPersistUpdateDataset	2✔
101	try {	1✔
102	lockStatus = metadataLock.lockWithRetries()	2✔
103	if (lockStatus) {	2✔
104	logInfo("Successfully able to get the table metadata file lock")	2✔
105	}
106	else {
107	throw new Exception("Table is locked for updation. Please try after some time")	×
108	}
109	// Get RDD.
110
111	dataSet = if (isPersistEnabled) {	2✔
112	Dataset.ofRows(sparkSession, plan).persist(StorageLevel.fromString(	2✔
113	CarbonProperties.getInstance.getUpdateDatasetStorageLevel()))	2✔
114	}
115	else {
116	Dataset.ofRows(sparkSession, plan)	1✔
117	}
118	val executionErrors = new ExecutionErrors(FailureCauses.NONE, "")	2✔
119
120
121	// handle the clean up of IUD.
122	CarbonUpdateUtil.cleanUpDeltaFiles(carbonTable, false)	2✔
123
124	// do delete operation.
125	val segmentsToBeDeleted = DeleteExecution.deleteDeltaExecution(	2✔
126	databaseNameOp,	2✔
127	tableName,	2✔
128	sparkSession,	2✔
129	dataSet.rdd,	2✔
130	currentTime + "",	2✔
131	isUpdateOperation = true,	2✔
132	executionErrors)	2✔
133
134	if (executionErrors.failureCauses != FailureCauses.NONE) {	2✔
135	throw new Exception(executionErrors.errorMsg)	×
136	}
137
138	// do update operation.
139	performUpdate(dataSet,	2✔
140	databaseNameOp,	2✔
141	tableName,	2✔
142	plan,	2✔
143	sparkSession,	2✔
144	currentTime,	2✔
145	executionErrors,	2✔
146	segmentsToBeDeleted)	2✔
147
148	if (executionErrors.failureCauses != FailureCauses.NONE) {	2✔
149	throw new Exception(executionErrors.errorMsg)	1✔
150	}
151
152	// Do IUD Compaction.
153	HorizontalCompaction.tryHorizontalCompaction(	2✔
154	sparkSession, carbonTable, isUpdateOperation = true)	2✔
155
156	// trigger event for Update table
157	val updateTablePostEvent: UpdateTablePostEvent =	2✔
158	UpdateTablePostEvent(sparkSession, carbonTable)	2✔
159	OperationListenerBus.getInstance.fireEvent(updateTablePostEvent, operationContext)	2✔
160	} catch {
161	case e: HorizontalCompactionException =>
162	LOGGER.error(	×
163	"Update operation passed. Exception in Horizontal Compaction. Please check logs." + e)	×
164	// In case of failure , clean all related delta files
165	CarbonUpdateUtil.cleanStaleDeltaFiles(carbonTable, e.compactionTimeStamp.toString)	×
166
167	case e: Exception =>
168	LOGGER.error("Exception in update operation", e)	1✔
169	// ****** start clean up.
170	// In case of failure , clean all related delete delta files
171	CarbonUpdateUtil.cleanStaleDeltaFiles(carbonTable, currentTime + "")	1✔
172
173	// *****end clean up.
174	if (null != e.getMessage) {	×
175	sys.error("Update operation failed. " + e.getMessage)	×
176	}
177	if (null != e.getCause && null != e.getCause.getMessage) {	×
178	sys.error("Update operation failed. " + e.getCause.getMessage)	×
179	}
180	sys.error("Update operation failed. please check logs.")	×
181	}
182	finally {	1✔
183	if (null != dataSet && isPersistEnabled) {	2✔
184	dataSet.unpersist()	2✔
185	}
186	if (lockStatus) {	2✔
187	CarbonLockUtil.fileUnlock(metadataLock, LockUsage.METADATA_LOCK)	2✔
188	}
189	}
190	Seq.empty	2✔
191	}
192
193	private def performUpdate(
194	dataFrame: Dataset[Row],
195	databaseNameOp: Option[String],
196	tableName: String,
197	plan: LogicalPlan,
198	sparkSession: SparkSession,
199	currentTime: Long,
200	executorErrors: ExecutionErrors,
201	deletedSegments: Seq[Segment]): Unit = {
202
203	def isDestinationRelation(relation: CarbonDatasourceHadoopRelation): Boolean = {	2✔
204	val dbName = CarbonEnv.getDatabaseName(databaseNameOp)(sparkSession)	2✔
205	(databaseNameOp.isDefined &&	2✔
206	databaseNameOp.get == dbName &&	2✔
207	tableName == relation.identifier.getCarbonTableIdentifier.getTableName) \|\|	2✔
208	(tableName == relation.identifier.getCarbonTableIdentifier.getTableName)	2✔
209	}
210
211	// from the dataFrame schema iterate through all the column to be updated and
212	// check for the data type, if the data type is complex then throw exception
213	def checkForUnsupportedDataType(dataFrame: DataFrame): Unit = {
214	dataFrame.schema.foreach(col => {	2✔
215	// the new column to be updated will be appended with "-updatedColumn" suffix
216	if (col.name.endsWith(CarbonCommonConstants.UPDATED_COL_EXTENSION) &&	2✔
217	col.dataType.isInstanceOf[ArrayType]) {	2✔
218	throw new UnsupportedOperationException("Unsupported data type: Array")	1✔
219	}
220	})
221	}
222
223	def getHeader(relation: CarbonDatasourceHadoopRelation, plan: LogicalPlan): String = {
224	var header = ""	2✔
225	var found = false	2✔
226
227	plan match {	2✔
228	case Project(pList, _) if (!found) =>	2✔
229	found = true	2✔
230	header = pList	2✔
231	.filter(field => !field.name	2✔
232	.equalsIgnoreCase(CarbonCommonConstants.CARBON_IMPLICIT_COLUMN_TUPLEID))	2✔
233	.map(col => if (col.name.endsWith(CarbonCommonConstants.UPDATED_COL_EXTENSION)) {	2✔
234	col.name	2✔
235	.substring(0, col.name.lastIndexOf(CarbonCommonConstants.UPDATED_COL_EXTENSION))	2✔
236	}
237	else {
238	col.name	2✔
239	}).mkString(",")	2✔
240	}
241	header	2✔
242	}
243
244	// check for the data type of the new value to be updated
245	checkForUnsupportedDataType(dataFrame)	2✔
246	val ex = dataFrame.queryExecution.analyzed	2✔
247	val res = ex find {	2✔
248	case relation: LogicalRelation	2✔
249	if relation.relation.isInstanceOf[CarbonDatasourceHadoopRelation] &&	2✔
250	isDestinationRelation(relation.relation.asInstanceOf[CarbonDatasourceHadoopRelation]) =>	2✔
251	true	2✔
252	case _ => false	2✔
253	}
254	val carbonRelation: CarbonDatasourceHadoopRelation = res match {	2✔
255	case Some(relation: LogicalRelation) =>	2✔
256	relation.relation.asInstanceOf[CarbonDatasourceHadoopRelation]	2✔
257	case _ => sys.error("")	×
258	}
259
260	val updateTableModel = UpdateTableModel(true, currentTime, executorErrors, deletedSegments)	2✔
261
262	val header = getHeader(carbonRelation, plan)	2✔
263
264	CarbonLoadDataCommand(	2✔
265	Some(carbonRelation.identifier.getCarbonTableIdentifier.getDatabaseName),	2✔
266	carbonRelation.identifier.getCarbonTableIdentifier.getTableName,	2✔
267	null,	2✔
268	Seq(),	2✔
269	Map(("fileheader" -> header)),	2✔
270	false,	2✔
271	null,	2✔
272	Some(dataFrame),	2✔
273	Some(updateTableModel)).run(sparkSession)	2✔
274
275	executorErrors.errorMsg = updateTableModel.executorErrors.errorMsg	2✔
276	executorErrors.failureCauses = updateTableModel.executorErrors.failureCauses	2✔
277
278	Seq.empty	2✔
279
280	}
281
282	override protected def opName: String = "UPDATE DATA"	2✔
283	}

apache / carbondata / 2475

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