Skip to content

Commit 2ee7775

Browse files
xuchuanyinjackylk
xuchuanyin
authored and
jackylk
committed
[CARBONDATA-1373] Enhance update performance by increasing parallelism
+ Increase parallelism while processing one segment in update + Use partitionBy instead of groupby + Return directly for no-rows-update case + Add a property to configure the parallelism + Clean up local files after update (previous bugs) This closes apache#1261
1 parent e8df8ba commit 2ee7775

File tree

7 files changed

+177
-58
lines changed

7 files changed

+177
-58
lines changed

‎core/src/main/java/org/apache/carbondata/core/constants/CarbonCommonConstants.java

+18
Original file line numberDiff line numberDiff line change
@@ -543,6 +543,10 @@ public final class CarbonCommonConstants {
543543
* UNDERSCORE
544544
*/
545545
public static final String UNDERSCORE = "_";
546+
/**
547+
* DASH
548+
*/
549+
public static final String DASH = "-";
546550
/**
547551
* POINT
548552
*/
@@ -1330,6 +1334,20 @@ public final class CarbonCommonConstants {
13301334
*/
13311335
public static final String CARBON_GLOBAL_SORT_RDD_STORAGE_LEVEL_DEFAULT = "MEMORY_ONLY";
13321336

1337+
/**
1338+
* property for configuring parallelism per segment when doing an update. Increase this
1339+
* value will avoid data screw problem for a large segment.
1340+
* Refer to CARBONDATA-1373 for more details.
1341+
*/
1342+
@CarbonProperty
1343+
public static final String CARBON_UPDATE_SEGMENT_PARALLELISM =
1344+
"carbon.update.segment.parallelism";
1345+
1346+
/**
1347+
* In default we will not optimize the update
1348+
*/
1349+
public static final String CARBON_UPDATE_SEGMENT_PARALLELISM_DEFAULT = "1";
1350+
13331351
private CarbonCommonConstants() {
13341352
}
13351353
}

‎core/src/main/java/org/apache/carbondata/core/util/CarbonProperties.java

+32
Original file line numberDiff line numberDiff line change
@@ -894,6 +894,38 @@ public String getGlobalSortRddStorageLevel() {
894894
return storageLevel.toUpperCase();
895895
}
896896

897+
/**
898+
* Returns parallelism for segment update
899+
* @return int
900+
*/
901+
public int getParallelismForSegmentUpdate() {
902+
int parallelism = Integer.parseInt(
903+
CarbonCommonConstants.CARBON_UPDATE_SEGMENT_PARALLELISM_DEFAULT);
904+
boolean isInvalidValue = false;
905+
try {
906+
String strParallelism = getProperty(CarbonCommonConstants.CARBON_UPDATE_SEGMENT_PARALLELISM,
907+
CarbonCommonConstants.CARBON_UPDATE_SEGMENT_PARALLELISM_DEFAULT);
908+
parallelism = Integer.parseInt(strParallelism);
909+
if (parallelism <= 0 || parallelism > 1000) {
910+
isInvalidValue = true;
911+
}
912+
} catch (NumberFormatException e) {
913+
isInvalidValue = true;
914+
}
915+
916+
if (isInvalidValue) {
917+
LOGGER.error("The specified value for property "
918+
+ CarbonCommonConstants.CARBON_UPDATE_SEGMENT_PARALLELISM
919+
+ " is incorrect. Correct value should be in range of 0 - 1000."
920+
+ " Taking the default value: "
921+
+ CarbonCommonConstants.CARBON_UPDATE_SEGMENT_PARALLELISM_DEFAULT);
922+
parallelism = Integer.parseInt(
923+
CarbonCommonConstants.CARBON_UPDATE_SEGMENT_PARALLELISM_DEFAULT);
924+
}
925+
926+
return parallelism;
927+
}
928+
897929
/**
898930
* returns true if carbon property
899931
* @param key

‎docs/configuration-parameters.md

+1-1
Original file line numberDiff line numberDiff line change
@@ -74,7 +74,7 @@ This section provides the details of all the configurations required for CarbonD
7474
| carbon.horizontal.compaction.enable | true | This property is used to turn ON/OFF horizontal compaction. After every DELETE and UPDATE statement, horizontal compaction may occur in case the delta (DELETE/ UPDATE) files becomes more than specified threshold. | |
7575
| carbon.horizontal.UPDATE.compaction.threshold | 1 | This property specifies the threshold limit on number of UPDATE delta files within a segment. In case the number of delta files goes beyond the threshold, the UPDATE delta files within the segment becomes eligible for horizontal compaction and compacted into single UPDATE delta file. | Values between 1 to 10000. |
7676
| carbon.horizontal.DELETE.compaction.threshold | 1 | This property specifies the threshold limit on number of DELETE delta files within a block of a segment. In case the number of delta files goes beyond the threshold, the DELETE delta files for the particular block of the segment becomes eligible for horizontal compaction and compacted into single DELETE delta file. | Values between 1 to 10000. |
77-
77+
| carbon.update.segment.parallelism | 1 | This property specifies the parallelism for each segment during update. If there are segments that contain too many records to update and the spark job encounter data-spill related errors, it is better to increase this property value. It is recommended to set this value to a multiple of the number of executors for balance. | Values between 1 to 1000. |
7878

7979

8080
* **Query Configuration**

‎integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/iud/UpdateCarbonTableTestCase.scala

+24
Original file line numberDiff line numberDiff line change
@@ -114,6 +114,30 @@ class UpdateCarbonTableTestCase extends QueryTest with BeforeAndAfterAll {
114114
sql("""drop table if exists iud.dest33""")
115115
}
116116

117+
test("update carbon table with optimized parallelism for segment") {
118+
sql("""drop table if exists iud.dest_opt_segment_parallelism""")
119+
sql(
120+
"""create table iud.dest_opt_segment_parallelism (c1 string,c2 int,c3 string,c5 string)
121+
| STORED BY 'org.apache.carbondata.format'""".stripMargin)
122+
sql(
123+
s"""LOAD DATA LOCAL INPATH '$resourcesPath/IUD/dest.csv'
124+
| INTO table iud.dest_opt_segment_parallelism""".stripMargin)
125+
sql(
126+
s"""LOAD DATA LOCAL INPATH '$resourcesPath/IUD/dest.csv'
127+
| INTO table iud.dest_opt_segment_parallelism""".stripMargin)
128+
CarbonProperties.getInstance().addProperty(
129+
CarbonCommonConstants.CARBON_UPDATE_SEGMENT_PARALLELISM, "3")
130+
sql(
131+
"""update iud.dest_opt_segment_parallelism d
132+
| set (c3,c5 ) = (select s.c33 ,s.c55 from iud.source2 s where d.c1 = s.c11)
133+
| where d.c1 = 'a'""".stripMargin).show()
134+
checkAnswer(
135+
sql("""select c3,c5 from iud.dest_opt_segment_parallelism where c1='a'"""),
136+
Seq(Row("MGM","Disco"),Row("MGM","Disco"))
137+
)
138+
sql("""drop table if exists iud.dest_opt_segment_parallelism""")
139+
}
140+
117141
test("update carbon table without alias in set three columns") {
118142
sql("""drop table if exists iud.dest44""")
119143
sql("""create table iud.dest44 (c1 string,c2 int,c3 string,c5 string) STORED BY 'org.apache.carbondata.format'""")

‎integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/UpdateDataLoad.scala

+3
Original file line numberDiff line numberDiff line change
@@ -28,6 +28,7 @@ import org.apache.carbondata.core.constants.CarbonCommonConstants
2828
import org.apache.carbondata.core.statusmanager.LoadMetadataDetails
2929
import org.apache.carbondata.processing.model.CarbonLoadModel
3030
import org.apache.carbondata.processing.newflow.DataLoadExecutor
31+
import org.apache.carbondata.spark.load.CarbonLoaderUtil
3132

3233
/**
3334
* Data load in case of update command .
@@ -62,6 +63,8 @@ object UpdateDataLoad {
6263
case e: Exception =>
6364
LOGGER.error(e)
6465
throw e
66+
} finally {
67+
CarbonLoaderUtil.deleteLocalDataLoadFolderLocation(carbonLoadModel, false, false)
6568
}
6669
}
6770

‎integration/spark/src/main/scala/org/apache/carbondata/spark/rdd/CarbonDataRDDFactory.scala

+49-28
Original file line numberDiff line numberDiff line change
@@ -19,7 +19,6 @@ package org.apache.carbondata.spark.rdd
1919

2020
import java.text.SimpleDateFormat
2121
import java.util
22-
import java.util.UUID
2322
import java.util.concurrent._
2423

2524
import scala.collection.JavaConverters._
@@ -32,8 +31,8 @@ import org.apache.hadoop.fs.Path
3231
import org.apache.hadoop.io.NullWritable
3332
import org.apache.hadoop.mapreduce.Job
3433
import org.apache.hadoop.mapreduce.lib.input.{FileInputFormat, FileSplit}
35-
import org.apache.spark.{SparkEnv, SparkException}
36-
import org.apache.spark.rdd.{DataLoadCoalescedRDD, DataLoadPartitionCoalescer, NewHadoopRDD, RDD, UpdateCoalescedRDD}
34+
import org.apache.spark.{SparkEnv, SparkException, TaskContext}
35+
import org.apache.spark.rdd.{DataLoadCoalescedRDD, DataLoadPartitionCoalescer, NewHadoopRDD, RDD}
3736
import org.apache.spark.sql.{CarbonEnv, DataFrame, Row, SQLContext}
3837
import org.apache.spark.sql.execution.command.{AlterTableModel, CompactionModel, ExecutionErrors, UpdateTableModel}
3938
import org.apache.spark.sql.hive.DistributionUtil
@@ -51,14 +50,14 @@ import org.apache.carbondata.core.metadata.schema.partition.PartitionType
5150
import org.apache.carbondata.core.metadata.schema.table.CarbonTable
5251
import org.apache.carbondata.core.mutate.CarbonUpdateUtil
5352
import org.apache.carbondata.core.scan.partition.PartitionUtil
54-
import org.apache.carbondata.core.statusmanager.LoadMetadataDetails
53+
import org.apache.carbondata.core.statusmanager.{LoadMetadataDetails, SegmentStatusManager}
5554
import org.apache.carbondata.core.util.{ByteUtil, CarbonProperties}
5655
import org.apache.carbondata.core.util.path.CarbonStorePath
5756
import org.apache.carbondata.processing.csvload.{BlockDetails, CSVInputFormat, StringArrayWritable}
5857
import org.apache.carbondata.processing.etl.DataLoadingException
5958
import org.apache.carbondata.processing.merger.{CarbonCompactionUtil, CarbonDataMergerUtil, CompactionType}
6059
import org.apache.carbondata.processing.model.CarbonLoadModel
61-
import org.apache.carbondata.processing.newflow.exception.{BadRecordFoundException, CarbonDataLoadingException}
60+
import org.apache.carbondata.processing.newflow.exception.BadRecordFoundException
6261
import org.apache.carbondata.processing.newflow.DataLoadProcessBuilder
6362
import org.apache.carbondata.processing.newflow.exception.CarbonDataLoadingException
6463
import org.apache.carbondata.processing.newflow.sort.SortScopeOptions
@@ -583,7 +582,9 @@ object CarbonDataRDDFactory {
583582
}
584583

585584
def loadDataFrameForUpdate(): Unit = {
586-
def triggerDataLoadForSegment(key: String,
585+
val segmentUpdateParallelism = CarbonProperties.getInstance().getParallelismForSegmentUpdate
586+
587+
def triggerDataLoadForSegment(key: String, taskNo: Int,
587588
iter: Iterator[Row]): Iterator[(String, (LoadMetadataDetails, ExecutionErrors))] = {
588589
val rddResult = new updateResultImpl()
589590
val LOGGER = LogServiceFactory.getLogService(this.getClass.getName)
@@ -594,11 +595,7 @@ object CarbonDataRDDFactory {
594595
var uniqueLoadStatusId = ""
595596
try {
596597
val segId = key
597-
val taskNo = CarbonUpdateUtil
598-
.getLatestTaskIdForSegment(segId,
599-
CarbonStorePath.getCarbonTablePath(carbonLoadModel.getStorePath,
600-
carbonTable.getCarbonTableIdentifier))
601-
val index = taskNo + 1
598+
val index = taskNo
602599
uniqueLoadStatusId = carbonLoadModel.getTableName +
603600
CarbonCommonConstants.UNDERSCORE +
604601
(index + "_0")
@@ -621,8 +618,6 @@ object CarbonDataRDDFactory {
621618

622619
// storeLocation = CarbonDataLoadRDD.initialize(carbonLoadModel, index)
623620
loadMetadataDetails.setLoadStatus(CarbonCommonConstants.STORE_LOADSTATUS_SUCCESS)
624-
val rddIteratorKey = CarbonCommonConstants.RDDUTIL_UPDATE_KEY +
625-
UUID.randomUUID().toString
626621
UpdateDataLoad.DataLoadForUpdate(segId,
627622
index,
628623
iter,
@@ -657,26 +652,52 @@ object CarbonDataRDDFactory {
657652

658653
val updateRdd = dataFrame.get.rdd
659654

655+
// return directly if no rows to update
656+
val noRowsToUpdate = updateRdd.isEmpty()
657+
if (noRowsToUpdate) {
658+
res = Array[List[(String, (LoadMetadataDetails, ExecutionErrors))]]()
659+
return
660+
}
660661

662+
// splitting as (key, value) i.e., (segment, updatedRows)
661663
val keyRDD = updateRdd.map(row =>
662-
// splitting as (key, value) i.e., (segment, updatedRows)
663-
(row.get(row.size - 1).toString, Row(row.toSeq.slice(0, row.size - 1): _*))
664-
)
665-
val groupBySegmentRdd = keyRDD.groupByKey()
664+
(row.get(row.size - 1).toString, Row(row.toSeq.slice(0, row.size - 1): _*)))
665+
666+
val loadMetadataDetails = SegmentStatusManager.readLoadMetadata(
667+
carbonTable.getMetaDataFilepath)
668+
val segmentIds = loadMetadataDetails.map(_.getLoadName)
669+
val segmentIdIndex = segmentIds.zipWithIndex.toMap
670+
val carbonTablePath = CarbonStorePath.getCarbonTablePath(carbonLoadModel.getStorePath,
671+
carbonTable.getCarbonTableIdentifier)
672+
val segmentId2maxTaskNo = segmentIds.map { segId =>
673+
(segId, CarbonUpdateUtil.getLatestTaskIdForSegment(segId, carbonTablePath))
674+
}.toMap
675+
676+
class SegmentPartitioner(segIdIndex: Map[String, Int], parallelism: Int)
677+
extends org.apache.spark.Partitioner {
678+
override def numPartitions: Int = segmentIdIndex.size * parallelism
679+
680+
override def getPartition(key: Any): Int = {
681+
val segId = key.asInstanceOf[String]
682+
// partitionId
683+
segmentIdIndex(segId) * parallelism + Random.nextInt(parallelism)
684+
}
685+
}
666686

667-
val nodeNumOfData = groupBySegmentRdd.partitions.flatMap[String, Array[String]] { p =>
668-
DataLoadPartitionCoalescer.getPreferredLocs(groupBySegmentRdd, p).map(_.host)
669-
}.distinct.size
670-
val nodes = DistributionUtil.ensureExecutorsByNumberAndGetNodeList(nodeNumOfData,
671-
sqlContext.sparkContext)
672-
val groupBySegmentAndNodeRdd =
673-
new UpdateCoalescedRDD[(String, scala.Iterable[Row])](groupBySegmentRdd,
674-
nodes.distinct.toArray)
687+
val partitionByRdd = keyRDD.partitionBy(new SegmentPartitioner(segmentIdIndex,
688+
segmentUpdateParallelism))
675689

676-
res = groupBySegmentAndNodeRdd.map(x =>
677-
triggerDataLoadForSegment(x._1, x._2.toIterator).toList
678-
).collect()
690+
// because partitionId=segmentIdIndex*parallelism+RandomPart and RandomPart<parallelism,
691+
// so segmentIdIndex=partitionId/parallelism, this has been verified.
692+
res = partitionByRdd.map(_._2).mapPartitions { partition =>
693+
val partitionId = TaskContext.getPartitionId()
694+
val segIdIndex = partitionId / segmentUpdateParallelism
695+
val randomPart = partitionId - segIdIndex * segmentUpdateParallelism
696+
val segId = segmentIds(segIdIndex)
697+
val newTaskNo = segmentId2maxTaskNo(segId) + randomPart + 1
679698

699+
List(triggerDataLoadForSegment(segId, newTaskNo, partition).toList).toIterator
700+
}.collect()
680701
}
681702

682703
def loadDataForPartitionTable(): Unit = {

‎integration/spark2/src/main/scala/org/apache/carbondata/spark/rdd/CarbonDataRDDFactory.scala

+50-29
Original file line numberDiff line numberDiff line change
@@ -19,7 +19,6 @@ package org.apache.carbondata.spark.rdd
1919

2020
import java.text.SimpleDateFormat
2121
import java.util
22-
import java.util.UUID
2322
import java.util.concurrent._
2423

2524
import scala.collection.JavaConverters._
@@ -32,10 +31,10 @@ import org.apache.hadoop.fs.Path
3231
import org.apache.hadoop.io.NullWritable
3332
import org.apache.hadoop.mapreduce.Job
3433
import org.apache.hadoop.mapreduce.lib.input.{FileInputFormat, FileSplit}
35-
import org.apache.spark.{SparkEnv, SparkException}
36-
import org.apache.spark.rdd.{DataLoadCoalescedRDD, DataLoadPartitionCoalescer, NewHadoopRDD, RDD, UpdateCoalescedRDD}
34+
import org.apache.spark.{SparkEnv, SparkException, TaskContext}
35+
import org.apache.spark.rdd.{DataLoadCoalescedRDD, DataLoadPartitionCoalescer, NewHadoopRDD, RDD}
3736
import org.apache.spark.sql.{CarbonEnv, DataFrame, Row, SQLContext}
38-
import org.apache.spark.sql.execution.command._
37+
import org.apache.spark.sql.execution.command.{AlterTableModel, CompactionModel, ExecutionErrors, UpdateTableModel}
3938
import org.apache.spark.sql.hive.DistributionUtil
4039
import org.apache.spark.util.SparkUtil
4140

@@ -59,8 +58,8 @@ import org.apache.carbondata.processing.csvload.{BlockDetails, CSVInputFormat, S
5958
import org.apache.carbondata.processing.etl.DataLoadingException
6059
import org.apache.carbondata.processing.merger.{CarbonCompactionUtil, CarbonDataMergerUtil, CompactionType}
6160
import org.apache.carbondata.processing.model.CarbonLoadModel
62-
import org.apache.carbondata.processing.newflow.DataLoadProcessBuilder
63-
import org.apache.carbondata.processing.newflow.exception.{BadRecordFoundException, CarbonDataLoadingException}
61+
import org.apache.carbondata.processing.newflow.exception.BadRecordFoundException
62+
import org.apache.carbondata.processing.newflow.exception.CarbonDataLoadingException
6463
import org.apache.carbondata.processing.newflow.sort.SortScopeOptions
6564
import org.apache.carbondata.processing.util.CarbonDataProcessorUtil
6665
import org.apache.carbondata.spark.{DataLoadResultImpl, PartitionFactory, _}
@@ -682,7 +681,9 @@ object CarbonDataRDDFactory {
682681
}
683682

684683
def loadDataFrameForUpdate(): Unit = {
685-
def triggerDataLoadForSegment(key: String,
684+
val segmentUpdateParallelism = CarbonProperties.getInstance().getParallelismForSegmentUpdate
685+
686+
def triggerDataLoadForSegment(key: String, taskNo: Int,
686687
iter: Iterator[Row]): Iterator[(String, (LoadMetadataDetails, ExecutionErrors))] = {
687688
val rddResult = new updateResultImpl()
688689
val LOGGER = LogServiceFactory.getLogService(this.getClass.getName)
@@ -693,11 +694,7 @@ object CarbonDataRDDFactory {
693694
var uniqueLoadStatusId = ""
694695
try {
695696
val segId = key
696-
val taskNo = CarbonUpdateUtil
697-
.getLatestTaskIdForSegment(segId,
698-
CarbonStorePath.getCarbonTablePath(carbonLoadModel.getStorePath,
699-
carbonTable.getCarbonTableIdentifier))
700-
val index = taskNo + 1
697+
val index = taskNo
701698
uniqueLoadStatusId = carbonLoadModel.getTableName +
702699
CarbonCommonConstants.UNDERSCORE +
703700
(index + "_0")
@@ -720,8 +717,6 @@ object CarbonDataRDDFactory {
720717

721718
// storeLocation = CarbonDataLoadRDD.initialize(carbonLoadModel, index)
722719
loadMetadataDetails.setLoadStatus(CarbonCommonConstants.STORE_LOADSTATUS_SUCCESS)
723-
val rddIteratorKey = CarbonCommonConstants.RDDUTIL_UPDATE_KEY +
724-
UUID.randomUUID().toString
725720
UpdateDataLoad.DataLoadForUpdate(segId,
726721
index,
727722
iter,
@@ -756,26 +751,52 @@ object CarbonDataRDDFactory {
756751

757752
val updateRdd = dataFrame.get.rdd
758753

754+
// return directly if no rows to update
755+
val noRowsToUpdate = updateRdd.isEmpty()
756+
if (noRowsToUpdate) {
757+
res = Array[List[(String, (LoadMetadataDetails, ExecutionErrors))]]()
758+
return
759+
}
759760

761+
// splitting as (key, value) i.e., (segment, updatedRows)
760762
val keyRDD = updateRdd.map(row =>
761-
// splitting as (key, value) i.e., (segment, updatedRows)
762-
(row.get(row.size - 1).toString, Row(row.toSeq.slice(0, row.size - 1): _*))
763-
)
764-
val groupBySegmentRdd = keyRDD.groupByKey()
763+
(row.get(row.size - 1).toString, Row(row.toSeq.slice(0, row.size - 1): _*)))
764+
765+
val loadMetadataDetails = SegmentStatusManager.readLoadMetadata(
766+
carbonTable.getMetaDataFilepath)
767+
val segmentIds = loadMetadataDetails.map(_.getLoadName)
768+
val segmentIdIndex = segmentIds.zipWithIndex.toMap
769+
val carbonTablePath = CarbonStorePath.getCarbonTablePath(carbonLoadModel.getStorePath,
770+
carbonTable.getCarbonTableIdentifier)
771+
val segmentId2maxTaskNo = segmentIds.map { segId =>
772+
(segId, CarbonUpdateUtil.getLatestTaskIdForSegment(segId, carbonTablePath))
773+
}.toMap
774+
775+
class SegmentPartitioner(segIdIndex: Map[String, Int], parallelism: Int)
776+
extends org.apache.spark.Partitioner {
777+
override def numPartitions: Int = segmentIdIndex.size * parallelism
778+
779+
override def getPartition(key: Any): Int = {
780+
val segId = key.asInstanceOf[String]
781+
// partitionId
782+
segmentIdIndex(segId) * parallelism + Random.nextInt(parallelism)
783+
}
784+
}
765785

766-
val nodeNumOfData = groupBySegmentRdd.partitions.flatMap[String, Array[String]] { p =>
767-
DataLoadPartitionCoalescer.getPreferredLocs(groupBySegmentRdd, p).map(_.host)
768-
}.distinct.size
769-
val nodes = DistributionUtil.ensureExecutorsByNumberAndGetNodeList(nodeNumOfData,
770-
sqlContext.sparkContext)
771-
val groupBySegmentAndNodeRdd =
772-
new UpdateCoalescedRDD[(String, scala.Iterable[Row])](groupBySegmentRdd,
773-
nodes.distinct.toArray)
786+
val partitionByRdd = keyRDD.partitionBy(new SegmentPartitioner(segmentIdIndex,
787+
segmentUpdateParallelism))
774788

775-
res = groupBySegmentAndNodeRdd.map(x =>
776-
triggerDataLoadForSegment(x._1, x._2.toIterator).toList
777-
).collect()
789+
// because partitionId=segmentIdIndex*parallelism+RandomPart and RandomPart<parallelism,
790+
// so segmentIdIndex=partitionId/parallelism, this has been verified.
791+
res = partitionByRdd.map(_._2).mapPartitions { partition =>
792+
val partitionId = TaskContext.getPartitionId()
793+
val segIdIndex = partitionId / segmentUpdateParallelism
794+
val randomPart = partitionId - segIdIndex * segmentUpdateParallelism
795+
val segId = segmentIds(segIdIndex)
796+
val newTaskNo = segmentId2maxTaskNo(segId) + randomPart + 1
778797

798+
List(triggerDataLoadForSegment(segId, newTaskNo, partition).toList).toIterator
799+
}.collect()
779800
}
780801

781802
def loadDataForPartitionTable(): Unit = {

0 commit comments

Comments
 (0)