You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@carbondata.apache.org by ra...@apache.org on 2018/07/30 18:42:34 UTC
[08/50] carbondata git commit: [CARBONDATA-2734] Update is not
working on the table which has segmentfile present
[CARBONDATA-2734] Update is not working on the table which has segmentfile present
It fixes the IUD on the flat folder
This closes #2503
Project: http://git-wip-us.apache.org/repos/asf/carbondata/repo
Commit: http://git-wip-us.apache.org/repos/asf/carbondata/commit/78a7371c
Tree: http://git-wip-us.apache.org/repos/asf/carbondata/tree/78a7371c
Diff: http://git-wip-us.apache.org/repos/asf/carbondata/diff/78a7371c
Branch: refs/heads/branch-1.4
Commit: 78a7371c4d55c69474c4bafda2c99c0536431dc4
Parents: c144e3d
Author: ravipesala <ra...@gmail.com>
Authored: Fri Jul 13 13:15:15 2018 +0530
Committer: ravipesala <ra...@gmail.com>
Committed: Tue Jul 31 00:10:41 2018 +0530
----------------------------------------------------------------------
.../core/mutate/CarbonUpdateUtil.java | 4 ++--
.../executor/impl/AbstractQueryExecutor.java | 5 +++--
.../SegmentUpdateStatusManager.java | 12 +++++------
.../apache/carbondata/core/util/CarbonUtil.java | 19 ++++++++++++++---
.../FlatFolderTableLoadingTestCase.scala | 21 +++++++++++++++++++
.../iud/DeleteCarbonTableTestCase.scala | 22 +++++++++++---------
.../iud/UpdateCarbonTableTestCase.scala | 13 ++++++++++++
.../command/mutation/DeleteExecution.scala | 14 +++++++------
8 files changed, 81 insertions(+), 29 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/carbondata/blob/78a7371c/core/src/main/java/org/apache/carbondata/core/mutate/CarbonUpdateUtil.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/mutate/CarbonUpdateUtil.java b/core/src/main/java/org/apache/carbondata/core/mutate/CarbonUpdateUtil.java
index d0a204c..4a8d2e8 100644
--- a/core/src/main/java/org/apache/carbondata/core/mutate/CarbonUpdateUtil.java
+++ b/core/src/main/java/org/apache/carbondata/core/mutate/CarbonUpdateUtil.java
@@ -81,10 +81,10 @@ public class CarbonUpdateUtil {
/**
* Returns block path from tuple id
*/
- public static String getTableBlockPath(String tid, String tablePath, boolean isPartitionTable) {
+ public static String getTableBlockPath(String tid, String tablePath, boolean isStandardTable) {
String partField = getRequiredFieldFromTID(tid, TupleIdEnum.PART_ID);
// If it has segment file then partfield can be appended directly to table path
- if (isPartitionTable) {
+ if (!isStandardTable) {
return tablePath + CarbonCommonConstants.FILE_SEPARATOR + partField.replace("#", "/");
}
String part = CarbonTablePath.addPartPrefix(partField);
http://git-wip-us.apache.org/repos/asf/carbondata/blob/78a7371c/core/src/main/java/org/apache/carbondata/core/scan/executor/impl/AbstractQueryExecutor.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/scan/executor/impl/AbstractQueryExecutor.java b/core/src/main/java/org/apache/carbondata/core/scan/executor/impl/AbstractQueryExecutor.java
index 180ca4d..910ae3e 100644
--- a/core/src/main/java/org/apache/carbondata/core/scan/executor/impl/AbstractQueryExecutor.java
+++ b/core/src/main/java/org/apache/carbondata/core/scan/executor/impl/AbstractQueryExecutor.java
@@ -344,11 +344,12 @@ public abstract class AbstractQueryExecutor<E> implements QueryExecutor<E> {
queryModel.getProjectionDimensions(), tableBlockDimensions,
segmentProperties.getComplexDimensions(), queryModel.getProjectionMeasures().size(),
queryModel.getTable().getTableInfo().isTransactionalTable());
+ boolean isStandardTable = CarbonUtil.isStandardCarbonTable(queryModel.getTable());
String blockId = CarbonUtil
.getBlockId(queryModel.getAbsoluteTableIdentifier(), filePath, segment.getSegmentNo(),
queryModel.getTable().getTableInfo().isTransactionalTable(),
- queryModel.getTable().isHivePartitionTable());
- if (queryModel.getTable().isHivePartitionTable()) {
+ isStandardTable);
+ if (!isStandardTable) {
blockExecutionInfo.setBlockId(CarbonTablePath.getShortBlockIdForPartitionTable(blockId));
} else {
blockExecutionInfo.setBlockId(CarbonTablePath.getShortBlockId(blockId));
http://git-wip-us.apache.org/repos/asf/carbondata/blob/78a7371c/core/src/main/java/org/apache/carbondata/core/statusmanager/SegmentUpdateStatusManager.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/statusmanager/SegmentUpdateStatusManager.java b/core/src/main/java/org/apache/carbondata/core/statusmanager/SegmentUpdateStatusManager.java
index 55381fb..0c2098a 100644
--- a/core/src/main/java/org/apache/carbondata/core/statusmanager/SegmentUpdateStatusManager.java
+++ b/core/src/main/java/org/apache/carbondata/core/statusmanager/SegmentUpdateStatusManager.java
@@ -69,21 +69,22 @@ public class SegmentUpdateStatusManager {
private LoadMetadataDetails[] segmentDetails;
private SegmentUpdateDetails[] updateDetails;
private Map<String, SegmentUpdateDetails> blockAndDetailsMap;
- private boolean isPartitionTable;
+ private boolean isStandardTable;
public SegmentUpdateStatusManager(CarbonTable table,
LoadMetadataDetails[] segmentDetails) {
this.identifier = table.getAbsoluteTableIdentifier();
+ this.isStandardTable = CarbonUtil.isStandardCarbonTable(table);
// current it is used only for read function scenarios, as file update always requires to work
// on latest file status.
this.segmentDetails = segmentDetails;
updateDetails = readLoadMetadata();
- isPartitionTable = table.isHivePartitionTable();
populateMap();
}
public SegmentUpdateStatusManager(CarbonTable table) {
this.identifier = table.getAbsoluteTableIdentifier();
+ this.isStandardTable = CarbonUtil.isStandardCarbonTable(table);
// current it is used only for read function scenarios, as file update always requires to work
// on latest file status.
if (!table.getTableInfo().isTransactionalTable()) {
@@ -98,7 +99,6 @@ public class SegmentUpdateStatusManager {
} else {
updateDetails = new SegmentUpdateDetails[0];
}
- isPartitionTable = table.isHivePartitionTable();
populateMap();
}
@@ -250,9 +250,9 @@ public class SegmentUpdateStatusManager {
*/
public String[] getDeleteDeltaFilePath(String blockFilePath, String segmentId) throws Exception {
String blockId =
- CarbonUtil.getBlockId(identifier, blockFilePath, segmentId, true, isPartitionTable);
+ CarbonUtil.getBlockId(identifier, blockFilePath, segmentId, true, isStandardTable);
String tupleId;
- if (isPartitionTable) {
+ if (!isStandardTable) {
tupleId = CarbonTablePath.getShortBlockIdForPartitionTable(blockId);
} else {
tupleId = CarbonTablePath.getShortBlockId(blockId);
@@ -272,7 +272,7 @@ public class SegmentUpdateStatusManager {
+ CarbonCommonConstants.FACT_FILE_EXT);
String blockPath;
- if (isPartitionTable) {
+ if (!isStandardTable) {
blockPath = identifier.getTablePath() + CarbonCommonConstants.FILE_SEPARATOR
+ CarbonUpdateUtil.getRequiredFieldFromTID(tupleId, TupleIdEnum.PART_ID)
.replace("#", "/") + CarbonCommonConstants.FILE_SEPARATOR + completeBlockName;
http://git-wip-us.apache.org/repos/asf/carbondata/blob/78a7371c/core/src/main/java/org/apache/carbondata/core/util/CarbonUtil.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/util/CarbonUtil.java b/core/src/main/java/org/apache/carbondata/core/util/CarbonUtil.java
index 642fe8e..3eb6aae 100644
--- a/core/src/main/java/org/apache/carbondata/core/util/CarbonUtil.java
+++ b/core/src/main/java/org/apache/carbondata/core/util/CarbonUtil.java
@@ -2906,17 +2906,17 @@ public final class CarbonUtil {
* @param identifier
* @param filePath
* @param segmentId
- * @param isTransactionalTable
+ * @param isStandardTable
* @return
*/
public static String getBlockId(AbsoluteTableIdentifier identifier, String filePath,
- String segmentId, boolean isTransactionalTable, boolean isPartitionTable) {
+ String segmentId, boolean isTransactionalTable, boolean isStandardTable) {
String blockId;
String blockName = filePath.substring(filePath.lastIndexOf("/") + 1, filePath.length());
String tablePath = identifier.getTablePath();
if (filePath.startsWith(tablePath)) {
- if (!isTransactionalTable || !isPartitionTable) {
+ if (!isTransactionalTable || isStandardTable) {
blockId = "Part0" + CarbonCommonConstants.FILE_SEPARATOR + "Segment_" + segmentId
+ CarbonCommonConstants.FILE_SEPARATOR + blockName;
} else {
@@ -3235,4 +3235,17 @@ public final class CarbonUtil {
int version = fileHeader.getVersion();
return ColumnarFormatVersion.valueOf((short)version);
}
+
+ /**
+ * Check whether it is standard table means tablepath has Fact/Part0/Segment_ tail present with
+ * all carbon files. In other cases carbon files present directly under tablepath or
+ * tablepath/partition folder
+ * TODO Read segment file and corresponding index file to get the correct carbondata file instead
+ * of using this way.
+ * @param table
+ * @return
+ */
+ public static boolean isStandardCarbonTable(CarbonTable table) {
+ return !(table.isSupportFlatFolder() || table.isHivePartitionTable());
+ }
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/78a7371c/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/flatfolder/FlatFolderTableLoadingTestCase.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/flatfolder/FlatFolderTableLoadingTestCase.scala b/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/flatfolder/FlatFolderTableLoadingTestCase.scala
index d786d10..9a60978 100644
--- a/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/flatfolder/FlatFolderTableLoadingTestCase.scala
+++ b/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/flatfolder/FlatFolderTableLoadingTestCase.scala
@@ -106,6 +106,27 @@ class FlatFolderTableLoadingTestCase extends QueryTest with BeforeAndAfterAll {
}
+ test("merge index flat folder issue") {
+ sql("drop table if exists t1")
+ sql("create table t1(c1 int,c2 string,c3 float,c4 date) stored by 'carbondata' TBLPROPERTIES('flat_folder'='true')")
+ sql("insert into t1 select 1,'a',1001,'1999-01-02'")
+ sql("insert into t1 select 2,'b',20.01,'1998-01-02'")
+ sql("insert into t1 select 3,'c',30.01,'1997-01-02'")
+ sql("insert into t1 select 4,'d',40.01,'1996-01-02'")
+ sql("insert into t1 select 5,'d',40.01,'1996-01-02'")
+ sql("delete from table t1 where segment.id in(1)")
+ val carbonTable = CarbonMetadata.getInstance().getCarbonTable("default", "t1")
+ assert(FileFactory.getCarbonFile(carbonTable.getTablePath).listFiles().filter(_.getName.endsWith(CarbonTablePath.MERGE_INDEX_FILE_EXT)).length == 5)
+ sql("clean files for table t1")
+ assert(FileFactory.getCarbonFile(carbonTable.getTablePath).listFiles().filter(_.getName.endsWith(CarbonTablePath.MERGE_INDEX_FILE_EXT)).length == 4)
+ sql("Alter table t1 compact 'minor'")
+ sql("show segments for table t1").show()
+ assert(FileFactory.getCarbonFile(carbonTable.getTablePath).listFiles().filter(_.getName.endsWith(CarbonTablePath.MERGE_INDEX_FILE_EXT)).length == 5)
+ sql("clean files for table t1")
+ assert(FileFactory.getCarbonFile(carbonTable.getTablePath).listFiles().filter(_.getName.endsWith(CarbonTablePath.MERGE_INDEX_FILE_EXT)).length == 1)
+ sql("drop table if exists t1")
+ }
+
override def afterAll = {
CarbonProperties.getInstance()
.addProperty(CarbonCommonConstants.CARBON_TIMESTAMP_FORMAT,
http://git-wip-us.apache.org/repos/asf/carbondata/blob/78a7371c/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/iud/DeleteCarbonTableTestCase.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/iud/DeleteCarbonTableTestCase.scala b/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/iud/DeleteCarbonTableTestCase.scala
index 8280693..39582f0 100644
--- a/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/iud/DeleteCarbonTableTestCase.scala
+++ b/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/iud/DeleteCarbonTableTestCase.scala
@@ -24,6 +24,7 @@ import org.apache.spark.sql.test.util.QueryTest
import org.apache.spark.sql.{CarbonEnv, Row, SaveMode}
import org.scalatest.BeforeAndAfterAll
+import org.apache.carbondata.core.datamap.Segment
import org.apache.carbondata.core.datastore.filesystem.{CarbonFile, CarbonFileFilter}
import org.apache.carbondata.core.datastore.impl.FileFactory
import org.apache.carbondata.core.mutate.CarbonUpdateUtil
@@ -246,15 +247,15 @@ class DeleteCarbonTableTestCase extends QueryTest with BeforeAndAfterAll {
val dataframe = sql("select getTupleId() as tupleId from iud_db.dest_tuple")
val listOfTupleId = dataframe.collect().map(df => df.get(0).toString).sorted
assert(
- listOfTupleId(0).startsWith("0/0/0-0_batchno0-0-0-") && listOfTupleId(0).endsWith("/0/0/0"))
+ listOfTupleId(0).contains("0/0/0-0_batchno0-0-0-") && listOfTupleId(0).endsWith("/0/0/0"))
assert(
- listOfTupleId(1).startsWith("0/0/0-0_batchno0-0-0-") && listOfTupleId(1).endsWith("/0/0/1"))
+ listOfTupleId(1).contains("0/0/0-0_batchno0-0-0-") && listOfTupleId(1).endsWith("/0/0/1"))
assert(
- listOfTupleId(2).startsWith("0/0/0-0_batchno0-0-0-") && listOfTupleId(2).endsWith("/0/0/2"))
+ listOfTupleId(2).contains("0/0/0-0_batchno0-0-0-") && listOfTupleId(2).endsWith("/0/0/2"))
assert(
- listOfTupleId(3).startsWith("0/0/0-0_batchno0-0-0-") && listOfTupleId(3).endsWith("/0/0/3"))
+ listOfTupleId(3).contains("0/0/0-0_batchno0-0-0-") && listOfTupleId(3).endsWith("/0/0/3"))
assert(
- listOfTupleId(4).startsWith("0/0/0-0_batchno0-0-0-") && listOfTupleId(4).endsWith("/0/0/4"))
+ listOfTupleId(4).contains("0/0/0-0_batchno0-0-0-") && listOfTupleId(4).endsWith("/0/0/4"))
val carbonTable_part = CarbonEnv.getInstance(Spark2TestQueryExecutor.spark).carbonMetastore
.lookupRelation(Option("iud_db"), "dest_tuple_part")(Spark2TestQueryExecutor.spark)
@@ -270,7 +271,7 @@ class DeleteCarbonTableTestCase extends QueryTest with BeforeAndAfterAll {
carbonDataFilename(0).getAbsolutePath,
"0",
carbonTable.isTransactionalTable,
- carbonTable.isHivePartitionTable)
+ CarbonUtil.isStandardCarbonTable(carbonTable))
assert(blockId.startsWith("Part0/Segment_0/part-0-0_batchno0-0-0-"))
val carbonDataFilename_part = new File(carbonTable_part.getTablePath + "/c3=aa").listFiles()
@@ -279,17 +280,18 @@ class DeleteCarbonTableTestCase extends QueryTest with BeforeAndAfterAll {
carbonDataFilename_part(0).getAbsolutePath,
"0",
carbonTable.isTransactionalTable,
- carbonTable.isHivePartitionTable)
+ CarbonUtil.isStandardCarbonTable(carbonTable))
assert(blockId_part.startsWith("Part0/Segment_0/part-0-100100000100001_batchno0-0-0-"))
-
+ val segment = Segment.getSegment("0", carbonTable.getTablePath)
val tableBlockPath = CarbonUpdateUtil
.getTableBlockPath(listOfTupleId(0),
carbonTable.getTablePath,
- carbonTable.isHivePartitionTable)
+ CarbonUtil.isStandardCarbonTable(carbonTable))
+ val segment_part = Segment.getSegment("0", carbonTable_part.getTablePath)
val tableBl0ckPath_part = CarbonUpdateUtil
.getTableBlockPath(listOfTupleId_part(0),
carbonTable_part.getTablePath,
- carbonTable_part.isHivePartitionTable)
+ CarbonUtil.isStandardCarbonTable(carbonTable_part))
assert(tableBl0ckPath_part.endsWith("iud_db.db/dest_tuple_part/c3=aa"))
assert(tableBlockPath.endsWith("iud_db.db/dest_tuple/Fact/Part0/Segment_0"))
http://git-wip-us.apache.org/repos/asf/carbondata/blob/78a7371c/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/iud/UpdateCarbonTableTestCase.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/iud/UpdateCarbonTableTestCase.scala b/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/iud/UpdateCarbonTableTestCase.scala
index 2cb2717..63a7b7b 100644
--- a/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/iud/UpdateCarbonTableTestCase.scala
+++ b/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/iud/UpdateCarbonTableTestCase.scala
@@ -759,6 +759,19 @@ class UpdateCarbonTableTestCase extends QueryTest with BeforeAndAfterAll {
sql("drop table if exists test_dm_index")
}
+ test("flat folder carbon table without alias in set columns with mulitple loads") {
+ sql("""drop table if exists iud.dest33_flat""")
+ sql("""create table iud.dest33_part (c1 string,c2 int,c5 string, c3 string) STORED BY 'org.apache.carbondata.format' TBLPROPERTIES('flat_folder'='true')""")
+ sql(s"""LOAD DATA LOCAL INPATH '$resourcesPath/IUD/dest.csv' INTO table iud.dest33_part""")
+ sql(s"""LOAD DATA LOCAL INPATH '$resourcesPath/IUD/dest.csv' INTO table iud.dest33_part""")
+ sql("""update iud.dest33_part d set (c3,c5 ) = (select s.c33 ,s.c55 from iud.source2 s where d.c1 = s.c11) where d.c1 = 'a'""").show()
+ checkAnswer(
+ sql("""select c3,c5 from iud.dest33_part where c1='a'"""),
+ Seq(Row("MGM","Disco"),Row("MGM","Disco"))
+ )
+ sql("""drop table if exists iud.dest33_part""")
+ }
+
override def afterAll {
sql("use default")
sql("drop database if exists iud cascade")
http://git-wip-us.apache.org/repos/asf/carbondata/blob/78a7371c/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/mutation/DeleteExecution.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/mutation/DeleteExecution.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/mutation/DeleteExecution.scala
index df3b961..8633243 100644
--- a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/mutation/DeleteExecution.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/mutation/DeleteExecution.scala
@@ -41,6 +41,7 @@ import org.apache.carbondata.core.metadata.AbsoluteTableIdentifier
import org.apache.carbondata.core.mutate.{CarbonUpdateUtil, DeleteDeltaBlockDetails, SegmentUpdateDetails, TupleIdEnum}
import org.apache.carbondata.core.mutate.data.RowCountDetailsVO
import org.apache.carbondata.core.statusmanager.{SegmentStatus, SegmentStatusManager, SegmentUpdateStatusManager}
+import org.apache.carbondata.core.util.CarbonUtil
import org.apache.carbondata.core.util.path.CarbonTablePath
import org.apache.carbondata.core.writer.CarbonDeleteDeltaWriterImpl
import org.apache.carbondata.hadoop.api.{CarbonInputFormat, CarbonTableInputFormat}
@@ -111,7 +112,7 @@ object DeleteExecution {
val metadataDetails = SegmentStatusManager.readTableStatusFile(
CarbonTablePath.getTableStatusFilePath(carbonTable.getTablePath))
-
+ val isStandardTable = CarbonUtil.isStandardCarbonTable(carbonTable)
val rowContRdd =
sparkSession.sparkContext.parallelize(
blockMappingVO.getCompleteBlockRowDetailVO.asScala.toSeq,
@@ -126,15 +127,15 @@ object DeleteExecution {
while (records.hasNext) {
val ((key), (rowCountDetailsVO, groupedRows)) = records.next
val segmentId = key.substring(0, key.indexOf(CarbonCommonConstants.FILE_SEPARATOR))
- val segmentFile =
- metadataDetails.find(_.getLoadName.equals(segmentId)).get.getSegmentFile
+ val loadDetail =
+ metadataDetails.find(_.getLoadName.equals(segmentId)).get
result = result ++
deleteDeltaFunc(index,
key,
groupedRows.toIterator,
timestamp,
rowCountDetailsVO,
- carbonTable.isHivePartitionTable)
+ isStandardTable)
}
result
}
@@ -222,7 +223,7 @@ object DeleteExecution {
iter: Iterator[Row],
timestamp: String,
rowCountDetailsVO: RowCountDetailsVO,
- isPartitionTable: Boolean
+ isStandardTable: Boolean
): Iterator[(SegmentStatus, (SegmentUpdateDetails, ExecutionErrors))] = {
val result = new DeleteDelataResultImpl()
@@ -258,7 +259,8 @@ object DeleteExecution {
countOfRows = countOfRows + 1
}
- val blockPath = CarbonUpdateUtil.getTableBlockPath(TID, tablePath, isPartitionTable)
+ val blockPath =
+ CarbonUpdateUtil.getTableBlockPath(TID, tablePath, isStandardTable)
val completeBlockName = CarbonTablePath
.addDataPartPrefix(CarbonUpdateUtil.getRequiredFieldFromTID(TID, TupleIdEnum.BLOCK_ID) +
CarbonCommonConstants.FACT_FILE_EXT)