You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@carbondata.apache.org by ak...@apache.org on 2021/06/07 16:38:21 UTC
[carbondata] branch master updated: [CARBONDATA-4193] Fix
compaction failure after alter add complex column.
This is an automated email from the ASF dual-hosted git repository.
akashrn5 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/carbondata.git
The following commit(s) were added to refs/heads/master by this push:
new fee8b18 [CARBONDATA-4193] Fix compaction failure after alter add complex column.
fee8b18 is described below
commit fee8b1822ecf238e9639fa51531e420181c1e065
Author: ShreelekhyaG <sh...@yahoo.com>
AuthorDate: Wed May 26 19:14:40 2021 +0530
[CARBONDATA-4193] Fix compaction failure after alter add complex column.
Why is this PR needed?
1. When we perform compaction after alter add a complex column, the query fails with
ArrayIndexOutOfBounds exception. While converting and adding row after merge step
in WriteStepRowUtil.fromMergerRow, As complex dimension is present, the complexKeys
array is accessed but doesnt have any values in array and throws exception.
2. Creating SI with globalsort on newly added complex column throws TreenodeException
(Caused by: java.lang.RuntimeException: Couldn't find positionId#172 in [arr2#153])
What changes were proposed in this PR?
1. While restructuring row, added changes to fill complexKeys with default values(null
values to children) according to the latest schema.
In SI queryresultprocessor, used the column property isParentColumnComplex to identify
any complex type. If complex index column not present in the parent table block,
assigned the SI row value to empty bytes.
2. For SI with globalsort, In case of complex type projection, TableProperties object in
carbonEnv is not same as in carbonTable object and hence requiredColumns is not
updated with positionId. So updating tableproperties from carbon env itself.
Does this PR introduce any user interface change?
No
Is any new testcase added?
Yes
This closes #4142
---
.../impl/RestructureBasedRawResultCollector.java | 63 +++++++++++++--
.../core/scan/executor/infos/DimensionInfo.java | 27 +++++++
.../core/scan/executor/util/RestructureUtil.java | 5 ++
.../apache/carbondata/core/util/CarbonUtil.java | 19 +++++
.../TestSIWithComplexArrayType.scala | 94 ++++++++++++++++++++--
.../query/SecondaryIndexQueryResultProcessor.java | 20 ++---
.../secondaryindex/command/SICreationCommand.scala | 8 +-
.../secondaryindex/rdd/SecondaryIndexCreator.scala | 11 +--
.../spark/src/test/resources/secindex/array2.csv | 4 +
.../alterTable/TestAlterTableAddColumns.scala | 18 +++++
.../processing/datatypes/PrimitiveDataType.java | 17 +---
11 files changed, 239 insertions(+), 47 deletions(-)
diff --git a/core/src/main/java/org/apache/carbondata/core/scan/collector/impl/RestructureBasedRawResultCollector.java b/core/src/main/java/org/apache/carbondata/core/scan/collector/impl/RestructureBasedRawResultCollector.java
index b2f5c07..80fb3c3 100644
--- a/core/src/main/java/org/apache/carbondata/core/scan/collector/impl/RestructureBasedRawResultCollector.java
+++ b/core/src/main/java/org/apache/carbondata/core/scan/collector/impl/RestructureBasedRawResultCollector.java
@@ -17,12 +17,18 @@
package org.apache.carbondata.core.scan.collector.impl;
+import java.io.ByteArrayOutputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
+import org.apache.carbondata.common.logging.LogServiceFactory;
import org.apache.carbondata.core.constants.CarbonCommonConstants;
+import org.apache.carbondata.core.metadata.datatype.DataType;
import org.apache.carbondata.core.metadata.datatype.DataTypes;
import org.apache.carbondata.core.metadata.encoder.Encoding;
+import org.apache.carbondata.core.metadata.schema.table.column.CarbonDimension;
import org.apache.carbondata.core.scan.executor.infos.BlockExecutionInfo;
import org.apache.carbondata.core.scan.model.ProjectionDimension;
import org.apache.carbondata.core.scan.model.ProjectionMeasure;
@@ -31,13 +37,19 @@ import org.apache.carbondata.core.scan.wrappers.ByteArrayWrapper;
import org.apache.carbondata.core.stats.QueryStatistic;
import org.apache.carbondata.core.stats.QueryStatisticsConstants;
import org.apache.carbondata.core.util.ByteUtil;
+import org.apache.carbondata.core.util.CarbonUtil;
import org.apache.carbondata.core.util.DataTypeUtil;
+import org.apache.log4j.Logger;
+
/**
* It is not a collector it is just a scanned result holder.
*/
public class RestructureBasedRawResultCollector extends RawBasedResultCollector {
+ private static final Logger LOGGER =
+ LogServiceFactory.getLogService(RestructureBasedRawResultCollector.class.getName());
+
public RestructureBasedRawResultCollector(BlockExecutionInfo blockExecutionInfos) {
super(blockExecutionInfos);
}
@@ -57,8 +69,9 @@ public class RestructureBasedRawResultCollector extends RawBasedResultCollector
if (dimensionInfo.isDictionaryColumnAdded()) {
fillDictionaryKeyArrayBatchWithLatestSchema(listBasedResult);
}
- if (dimensionInfo.isNoDictionaryColumnAdded()) {
- fillNoDictionaryKeyArrayBatchWithLatestSchema(listBasedResult);
+ if (dimensionInfo.isNoDictionaryColumnAdded() || dimensionInfo
+ .isComplexColumnAdded()) {
+ fillNoDictionaryAndComplexKeyArrayBatchWithLatestSchema(listBasedResult);
}
QueryStatistic resultPrepTime = queryStatisticsModel.getStatisticsTypeAndObjMap()
.get(QueryStatisticsConstants.RESULT_PREP_TIME);
@@ -111,36 +124,71 @@ public class RestructureBasedRawResultCollector extends RawBasedResultCollector
}
/**
- * This method will fill the no dictionary byte array with newly added no dictionary columns
+ * This method will fill the no dictionary and complex byte array with newly added columns
*
* @param rows
* @return
*/
- private void fillNoDictionaryKeyArrayBatchWithLatestSchema(List<Object[]> rows) {
+ private void fillNoDictionaryAndComplexKeyArrayBatchWithLatestSchema(List<Object[]> rows) {
for (Object[] row : rows) {
ByteArrayWrapper byteArrayWrapper = (ByteArrayWrapper) row[0];
byte[][] noDictKeyArray = byteArrayWrapper.getNoDictionaryKeys();
+ byte[][] complexTypesKeyArray = byteArrayWrapper.getComplexTypesKeys();
ProjectionDimension[] actualQueryDimensions = executionInfo.getActualQueryDimensions();
byte[][] noDictionaryKeyArrayWithNewlyAddedColumns =
new byte[noDictKeyArray.length + dimensionInfo.getNewNoDictionaryColumnCount()][];
+ byte[][] complexTypeKeyArrayWithNewlyAddedColumns =
+ new byte[complexTypesKeyArray.length + dimensionInfo.getNewComplexColumnCount()][];
int existingColumnValueIndex = 0;
int newKeyArrayIndex = 0;
+ int existingComplexColumnValueIndex = 0;
+ int newComplexKeyArrayIndex = 0;
for (int i = 0; i < dimensionInfo.getDimensionExists().length; i++) {
if (actualQueryDimensions[i].getDimension().getDataType() != DataTypes.DATE
&& !actualQueryDimensions[i].getDimension().hasEncoding(Encoding.IMPLICIT)) {
+ DataType currDataType = actualQueryDimensions[i].getDimension().getDataType();
// if dimension exists then add the byte array value else add the default value
if (dimensionInfo.getDimensionExists()[i]) {
- noDictionaryKeyArrayWithNewlyAddedColumns[newKeyArrayIndex++] =
- noDictKeyArray[existingColumnValueIndex++];
+ if (currDataType.isComplexType()) {
+ complexTypeKeyArrayWithNewlyAddedColumns[newComplexKeyArrayIndex++] =
+ complexTypesKeyArray[existingComplexColumnValueIndex++];
+ } else {
+ noDictionaryKeyArrayWithNewlyAddedColumns[newKeyArrayIndex++] =
+ noDictKeyArray[existingColumnValueIndex++];
+ }
} else {
byte[] newColumnDefaultValue = null;
Object defaultValue = dimensionInfo.getDefaultValues()[i];
if (null != defaultValue) {
newColumnDefaultValue = (byte[]) defaultValue;
- } else if (actualQueryDimensions[i].getDimension().getDataType() == DataTypes.STRING) {
+ } else if (currDataType == DataTypes.STRING) {
newColumnDefaultValue =
DataTypeUtil.getDataTypeConverter().convertFromByteToUTF8Bytes(
CarbonCommonConstants.MEMBER_DEFAULT_VAL_ARRAY);
+ } else if (currDataType.isComplexType()) {
+ // Iterate over child dimensions and add its default value.
+ List<CarbonDimension> children =
+ actualQueryDimensions[i].getDimension().getListOfChildDimensions();
+ try (ByteArrayOutputStream byteStream = new ByteArrayOutputStream();
+ DataOutputStream dataOutputStream = new DataOutputStream(byteStream)) {
+ if (DataTypes.isArrayType(currDataType)) {
+ dataOutputStream.writeInt(1);
+ } else if (DataTypes.isStructType(currDataType)) {
+ dataOutputStream.writeShort(children.size());
+ }
+ for (int j = 0; j < children.size(); j++) {
+ // update default null values based on datatype
+ CarbonUtil.updateNullValueBasedOnDatatype(dataOutputStream,
+ children.get(j).getDataType());
+ }
+ newColumnDefaultValue = byteStream.toByteArray();
+ } catch (IOException e) {
+ LOGGER.error(e.getMessage(), e);
+ throw new RuntimeException(e);
+ }
+ complexTypeKeyArrayWithNewlyAddedColumns[newComplexKeyArrayIndex++] =
+ newColumnDefaultValue;
+ continue;
} else {
newColumnDefaultValue = CarbonCommonConstants.EMPTY_BYTE_ARRAY;
}
@@ -149,6 +197,7 @@ public class RestructureBasedRawResultCollector extends RawBasedResultCollector
}
}
byteArrayWrapper.setNoDictionaryKeys(noDictionaryKeyArrayWithNewlyAddedColumns);
+ byteArrayWrapper.setComplexTypesKeys(complexTypeKeyArrayWithNewlyAddedColumns);
}
}
}
diff --git a/core/src/main/java/org/apache/carbondata/core/scan/executor/infos/DimensionInfo.java b/core/src/main/java/org/apache/carbondata/core/scan/executor/infos/DimensionInfo.java
index b41de82..0a9820f 100644
--- a/core/src/main/java/org/apache/carbondata/core/scan/executor/infos/DimensionInfo.java
+++ b/core/src/main/java/org/apache/carbondata/core/scan/executor/infos/DimensionInfo.java
@@ -56,6 +56,17 @@ public class DimensionInfo {
* count of no dictionary columns not existing in the current block
*/
private int newNoDictionaryColumnCount;
+
+ /**
+ * count of complex columns not existing in the current block
+ */
+ private int newComplexColumnCount;
+
+ /**
+ * flag to check whether there exist a complex column in the query which
+ * does not exist in the current block
+ */
+ private boolean isComplexColumnAdded;
/**
* maintains the block datatype
*/
@@ -115,4 +126,20 @@ public class DimensionInfo {
public void setNewNoDictionaryColumnCount(int newNoDictionaryColumnCount) {
this.newNoDictionaryColumnCount = newNoDictionaryColumnCount;
}
+
+ public boolean isComplexColumnAdded() {
+ return isComplexColumnAdded;
+ }
+
+ public void setComplexColumnAdded(boolean complexColumnAdded) {
+ isComplexColumnAdded = complexColumnAdded;
+ }
+
+ public int getNewComplexColumnCount() {
+ return newComplexColumnCount;
+ }
+
+ public void setNewComplexColumnCount(int newComplexColumnCount) {
+ this.newComplexColumnCount = newComplexColumnCount;
+ }
}
diff --git a/core/src/main/java/org/apache/carbondata/core/scan/executor/util/RestructureUtil.java b/core/src/main/java/org/apache/carbondata/core/scan/executor/util/RestructureUtil.java
index 2f2cdfe..7fabe17 100644
--- a/core/src/main/java/org/apache/carbondata/core/scan/executor/util/RestructureUtil.java
+++ b/core/src/main/java/org/apache/carbondata/core/scan/executor/util/RestructureUtil.java
@@ -77,6 +77,7 @@ public class RestructureUtil {
dimensionInfo.dataType = new DataType[queryDimensions.length + measureCount];
int newDictionaryColumnCount = 0;
int newNoDictionaryColumnCount = 0;
+ int newNoDictionaryComplexColumnCount = 0;
// selecting only those dimension which is present in the query
int dimIndex = 0;
for (ProjectionDimension queryDimension : queryDimensions) {
@@ -140,6 +141,9 @@ public class RestructureUtil {
if (queryDimension.getDimension().getDataType() == DataTypes.DATE) {
dimensionInfo.setDictionaryColumnAdded(true);
newDictionaryColumnCount++;
+ } else if (queryDimension.getDimension().getDataType().isComplexType()) {
+ dimensionInfo.setComplexColumnAdded(true);
+ newNoDictionaryComplexColumnCount++;
} else {
dimensionInfo.setNoDictionaryColumnAdded(true);
newNoDictionaryColumnCount++;
@@ -150,6 +154,7 @@ public class RestructureUtil {
}
dimensionInfo.setNewDictionaryColumnCount(newDictionaryColumnCount);
dimensionInfo.setNewNoDictionaryColumnCount(newNoDictionaryColumnCount);
+ dimensionInfo.setNewComplexColumnCount(newNoDictionaryComplexColumnCount);
blockExecutionInfo.setDimensionInfo(dimensionInfo);
return presentDimension;
}
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 7b762ba..1b81950 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
@@ -3481,4 +3481,23 @@ public final class CarbonUtil {
});
}
}
+
+ public static void updateNullValueBasedOnDatatype(DataOutputStream dataOutputStream,
+ DataType dataType) throws IOException {
+ if (dataType == DataTypes.STRING || dataType == DataTypes.VARCHAR) {
+ if (DataTypeUtil.isByteArrayComplexChildColumn(dataType)) {
+ dataOutputStream.writeInt(CarbonCommonConstants.MEMBER_DEFAULT_VAL_ARRAY.length);
+ } else {
+ dataOutputStream.writeShort(CarbonCommonConstants.MEMBER_DEFAULT_VAL_ARRAY.length);
+ }
+ dataOutputStream.write(CarbonCommonConstants.MEMBER_DEFAULT_VAL_ARRAY);
+ } else {
+ if (DataTypeUtil.isByteArrayComplexChildColumn(dataType)) {
+ dataOutputStream.writeInt(CarbonCommonConstants.EMPTY_BYTE_ARRAY.length);
+ } else {
+ dataOutputStream.writeShort(CarbonCommonConstants.EMPTY_BYTE_ARRAY.length);
+ }
+ dataOutputStream.write(CarbonCommonConstants.EMPTY_BYTE_ARRAY);
+ }
+ }
}
diff --git a/index/secondary-index/src/test/scala/org/apache/carbondata/spark/testsuite/secondaryindex/TestSIWithComplexArrayType.scala b/index/secondary-index/src/test/scala/org/apache/carbondata/spark/testsuite/secondaryindex/TestSIWithComplexArrayType.scala
index 9e499e7..ef8bf3d 100644
--- a/index/secondary-index/src/test/scala/org/apache/carbondata/spark/testsuite/secondaryindex/TestSIWithComplexArrayType.scala
+++ b/index/secondary-index/src/test/scala/org/apache/carbondata/spark/testsuite/secondaryindex/TestSIWithComplexArrayType.scala
@@ -45,7 +45,7 @@ class TestSIWithComplexArrayType extends QueryTest with BeforeAndAfterEach {
sql("drop table if exists complextable5")
}
- test("Test alter add array column before creating SI") {
+ test("Test restructured array<string> and existing string column as index columns on SI with compaction") {
sql("drop table if exists complextable")
sql("create table complextable (id string, country array<string>, name string) stored as carbondata")
sql("insert into complextable select 1,array('china', 'us'), 'b'")
@@ -60,11 +60,12 @@ class TestSIWithComplexArrayType extends QueryTest with BeforeAndAfterEach {
checkAnswer(sql("select * from complextable where array_contains(arr2,'iron')"),
Seq(Row("4", mutable.WrappedArray.make(Array("India")), "g",
mutable.WrappedArray.make(Array("iron", "man", "jarvis")))))
- val result1 = sql("select * from complextable where array_contains(arr2,'iron')")
- val result2 = sql("select * from complextable where arr2[0]='iron'")
- sql("create index index_11 on table complextable(arr2) as 'carbondata'")
- val df1 = sql(" select * from complextable where array_contains(arr2,'iron')")
- val df2 = sql(" select * from complextable where arr2[0]='iron'")
+ val result1 = sql("select * from complextable where array_contains(arr2,'iron') and name='g'")
+ val result2 = sql("select * from complextable where arr2[0]='iron' and name='f'")
+ sql("create index index_11 on table complextable(arr2, name) as 'carbondata'")
+ sql("alter table complextable compact 'minor'")
+ val df1 = sql(" select * from complextable where array_contains(arr2,'iron') and name='g'")
+ val df2 = sql(" select * from complextable where arr2[0]='iron' and name='f'")
if (!isFilterPushedDownToSI(df1.queryExecution.sparkPlan)) {
assert(false)
} else {
@@ -85,7 +86,49 @@ class TestSIWithComplexArrayType extends QueryTest with BeforeAndAfterEach {
checkAnswer(result2, df2)
}
- test("test array<string> on secondary index") {
+ test("Test restructured array<string> and string columns as index columns on SI with compaction") {
+ sql("drop table if exists complextable")
+ sql("create table complextable (id string, country array<string>, name string) stored as carbondata")
+ sql("insert into complextable select 1,array('china', 'us'), 'b'")
+ sql("insert into complextable select 2,array('pak'), 'v'")
+
+ sql("drop index if exists index_11 on complextable")
+ sql(
+ "ALTER TABLE complextable ADD COLUMNS(arr2 array<string>)")
+ sql("ALTER TABLE complextable ADD COLUMNS(addr string)")
+ sql("insert into complextable select 3,array('china'), 'f',array('hello','world'),'china'")
+ sql("insert into complextable select 4,array('India'),'g',array('iron','man','jarvis'),'India'")
+
+ checkAnswer(sql("select * from complextable where array_contains(arr2,'iron')"),
+ Seq(Row("4", mutable.WrappedArray.make(Array("India")), "g",
+ mutable.WrappedArray.make(Array("iron", "man", "jarvis")), "India")))
+ val result1 = sql("select * from complextable where array_contains(arr2,'iron') and addr='India'")
+ val result2 = sql("select * from complextable where arr2[0]='iron' and addr='china'")
+ sql("create index index_11 on table complextable(arr2, addr) as 'carbondata'")
+ sql("alter table complextable compact 'minor'")
+ val df1 = sql(" select * from complextable where array_contains(arr2,'iron') and addr='India'")
+ val df2 = sql(" select * from complextable where arr2[0]='iron' and addr='china'")
+ if (!isFilterPushedDownToSI(df1.queryExecution.sparkPlan)) {
+ assert(false)
+ } else {
+ assert(true)
+ }
+ if (!isFilterPushedDownToSI(df2.queryExecution.sparkPlan)) {
+ assert(false)
+ } else {
+ assert(true)
+ }
+ val doNotHitSIDf = sql(" select * from complextable where array_contains(arr2,'iron') and array_contains(arr2,'man')")
+ if (isFilterPushedDownToSI(doNotHitSIDf.queryExecution.sparkPlan)) {
+ assert(false)
+ } else {
+ assert(true)
+ }
+ checkAnswer(result1, df1)
+ checkAnswer(result2, df2)
+ }
+
+ test("test array<string> on secondary index with compaction") {
sql("create table complextable (id string, country array<string>, name string) stored as carbondata")
sql("insert into complextable select 1,array('china', 'us'), 'b'")
sql("insert into complextable select 2,array('pak'), 'v'")
@@ -95,6 +138,7 @@ class TestSIWithComplexArrayType extends QueryTest with BeforeAndAfterEach {
val result2 = sql(" select * from complextable where country[0]='china'")
sql("drop index if exists index_1 on complextable")
sql("create index index_1 on table complextable(country) as 'carbondata'")
+ sql("alter table complextable compact 'minor'")
val df1 = sql(" select * from complextable where array_contains(country,'china')")
val df2 = sql(" select * from complextable where country[0]='china'")
if (!isFilterPushedDownToSI(df1.queryExecution.sparkPlan)) {
@@ -117,7 +161,7 @@ class TestSIWithComplexArrayType extends QueryTest with BeforeAndAfterEach {
checkAnswer(result2, df2)
}
- test("test array<string> and string as index columns on secondary index") {
+ test("test array<string> and string as index columns on secondary index with compaction") {
sql("create table complextable (id string, country array<string>, name string) stored as carbondata")
sql("insert into complextable select 1, array('china', 'us'), 'b'")
sql("insert into complextable select 2, array('pak'), 'v'")
@@ -126,6 +170,7 @@ class TestSIWithComplexArrayType extends QueryTest with BeforeAndAfterEach {
val result = sql(" select * from complextable where array_contains(country,'china') and name='f'")
sql("drop index if exists index_1 on complextable")
sql("create index index_1 on table complextable(country, name) as 'carbondata'")
+ sql("alter table complextable compact 'minor'")
val df = sql(" select * from complextable where array_contains(country,'china') and name='f'")
if (!isFilterPushedDownToSI(df.queryExecution.sparkPlan)) {
assert(false)
@@ -177,6 +222,39 @@ class TestSIWithComplexArrayType extends QueryTest with BeforeAndAfterEach {
.addProperty(CarbonCommonConstants.CARBON_SI_SEGMENT_MERGE, "false")
}
+ test("test SI global sort with si segment merge enabled for newly added complex column") {
+ CarbonProperties.getInstance()
+ .addProperty(CarbonCommonConstants.CARBON_SI_SEGMENT_MERGE, "true")
+ sql("create table complextable2 (id int, name string, country array<string>) stored as " +
+ "carbondata tblproperties('sort_scope'='global_sort','sort_columns'='name')")
+ sql(
+ s"load data inpath '$resourcesPath/secindex/array.csv' into table complextable2 options" +
+ s"('delimiter'=','," +
+ "'quotechar'='\"','fileheader'='id,name,country','complex_delimiter_level_1'='$'," +
+ "'global_sort_partitions'='10')")
+ sql("ALTER TABLE complextable2 ADD COLUMNS(arr2 array<string>)")
+ sql(
+ s"load data inpath '$resourcesPath/secindex/array2.csv' into table complextable2 options" +
+ s"('delimiter'=','," +
+ "'quotechar'='\"','fileheader'='id,name,country,arr2','complex_delimiter_level_1'='$'," +
+ "'global_sort_partitions'='10')")
+ val result = sql(" select * from complextable2 where array_contains(arr2,'iron')")
+ sql("drop index if exists index_2 on complextable")
+ sql("create index index_2 on table complextable2(arr2) as 'carbondata' properties" +
+ "('sort_scope'='global_sort') ")
+ checkAnswer(sql("select count(*) from complextable2 where array_contains(arr2,'iron')"),
+ sql("select count(*) from complextable2 where ni(array_contains(arr2,'iron'))"))
+ val df = sql(" select * from complextable2 where array_contains(arr2,'iron')")
+ if (!isFilterPushedDownToSI(df.queryExecution.sparkPlan)) {
+ assert(false)
+ } else {
+ assert(true)
+ }
+ checkAnswer(result, df)
+ CarbonProperties.getInstance()
+ .addProperty(CarbonCommonConstants.CARBON_SI_SEGMENT_MERGE, "false")
+ }
+
test("test SI global sort with si segment merge enabled for primitive data types") {
CarbonProperties.getInstance()
.addProperty(CarbonCommonConstants.CARBON_SI_SEGMENT_MERGE, "true")
diff --git a/integration/spark/src/main/java/org/apache/spark/sql/secondaryindex/query/SecondaryIndexQueryResultProcessor.java b/integration/spark/src/main/java/org/apache/spark/sql/secondaryindex/query/SecondaryIndexQueryResultProcessor.java
index 41a5c43..d8ff50f 100644
--- a/integration/spark/src/main/java/org/apache/spark/sql/secondaryindex/query/SecondaryIndexQueryResultProcessor.java
+++ b/integration/spark/src/main/java/org/apache/spark/sql/secondaryindex/query/SecondaryIndexQueryResultProcessor.java
@@ -277,15 +277,11 @@ public class SecondaryIndexQueryResultProcessor {
CarbonDimension dims = dimensions.get(i);
boolean isComplexColumn = false;
// As complex column of MainTable is stored as its primitive type in SI,
- // we need to check if dimension is complex dimension or not based on dimension
- // name. Check if name exists in complexDimensionInfoMap of main table result
- if (!complexDimensionInfoMap.isEmpty() && complexColumnParentBlockIndexes.length > 0) {
- for (GenericQueryType queryType : complexDimensionInfoMap.values()) {
- if (queryType.getName().equalsIgnoreCase(dims.getColName())) {
- isComplexColumn = true;
- break;
- }
- }
+ // we need to check if dimension is complex dimension or not based on isParentColumnComplex
+ // property.
+ if (dims.getColumnProperties() != null && Boolean
+ .parseBoolean(dims.getColumnProperties().get("isParentColumnComplex"))) {
+ isComplexColumn = true;
}
// fill all the no dictionary and dictionary data to the prepared row first, fill the complex
// flatten data to prepared row at last
@@ -293,7 +289,11 @@ public class SecondaryIndexQueryResultProcessor {
// dictionary
preparedRow[i] = wrapper.getDictionaryKeyByIndex(dictionaryIndex++);
} else {
- if (isComplexColumn) {
+ if (isComplexColumn && complexColumnParentBlockIndexes.length == 0) {
+ // After restructure some complex column will not be present in parent block.
+ // In such case, set the SI implicit row value to empty byte array.
+ preparedRow[i] = new byte[0];
+ } else if (isComplexColumn) {
// get the flattened data of complex column
byte[] complexKeyByIndex = wrapper.getComplexKeyByIndex(complexIndex);
ByteBuffer byteArrayInput = ByteBuffer.wrap(complexKeyByIndex);
diff --git a/integration/spark/src/main/scala/org/apache/spark/sql/secondaryindex/command/SICreationCommand.scala b/integration/spark/src/main/scala/org/apache/spark/sql/secondaryindex/command/SICreationCommand.scala
index b77677a..223bff1 100644
--- a/integration/spark/src/main/scala/org/apache/spark/sql/secondaryindex/command/SICreationCommand.scala
+++ b/integration/spark/src/main/scala/org/apache/spark/sql/secondaryindex/command/SICreationCommand.scala
@@ -680,18 +680,24 @@ private[sql] case class CarbonCreateSecondaryIndexCommand(
dataType: DataType = null): ColumnSchema = {
val columnSchema = new ColumnSchema()
val encodingList = parentColumnSchema.getEncodingList
+ var colPropMap = parentColumnSchema.getColumnProperties
// if data type is arrayType, then store the column as its CHILD data type in SI
if (DataTypes.isArrayType(parentColumnSchema.getDataType)) {
columnSchema.setDataType(dataType)
+ if (colPropMap == null) {
+ colPropMap = new java.util.HashMap[String, String]()
+ }
+ colPropMap.put("isParentColumnComplex", "true")
+ columnSchema.setColumnProperties(colPropMap)
if (dataType == DataTypes.DATE) {
encodingList.add(Encoding.DIRECT_DICTIONARY)
encodingList.add(Encoding.DICTIONARY)
}
} else {
columnSchema.setDataType(parentColumnSchema.getDataType)
+ columnSchema.setColumnProperties(colPropMap)
}
columnSchema.setColumnName(parentColumnSchema.getColumnName)
- columnSchema.setColumnProperties(parentColumnSchema.getColumnProperties)
columnSchema.setEncodingList(encodingList)
columnSchema.setColumnUniqueId(parentColumnSchema.getColumnUniqueId)
columnSchema.setColumnReferenceId(parentColumnSchema.getColumnReferenceId)
diff --git a/integration/spark/src/main/scala/org/apache/spark/sql/secondaryindex/rdd/SecondaryIndexCreator.scala b/integration/spark/src/main/scala/org/apache/spark/sql/secondaryindex/rdd/SecondaryIndexCreator.scala
index bc96996..eafbb3a 100644
--- a/integration/spark/src/main/scala/org/apache/spark/sql/secondaryindex/rdd/SecondaryIndexCreator.scala
+++ b/integration/spark/src/main/scala/org/apache/spark/sql/secondaryindex/rdd/SecondaryIndexCreator.scala
@@ -195,7 +195,7 @@ object SecondaryIndexCreator {
mainTable,
projections.mkString(","),
Array(eachSegment),
- isPositionReferenceRequired = true)
+ isPositionReferenceRequired = true, !explodeColumn.isEmpty)
// flatten the complex SI
if (explodeColumn.nonEmpty) {
val columns = dataFrame.schema.map { x =>
@@ -564,7 +564,8 @@ object SecondaryIndexCreator {
carbonTable: CarbonTable,
projections: String,
segments: Array[String],
- isPositionReferenceRequired: Boolean = false): DataFrame = {
+ isPositionReferenceRequired: Boolean = false,
+ isComplexTypeProjection: Boolean = false): DataFrame = {
try {
CarbonThreadUtil.threadSet(
CarbonCommonConstants.CARBON_INPUT_SEGMENTS + carbonTable.getDatabaseName +
@@ -584,9 +585,9 @@ object SecondaryIndexCreator {
case p: Project =>
Project(p.projectList :+ positionId, p.child)
}
- val tableProperties = if (carbonTable.isHivePartitionTable) {
- // in case of partition table, TableProperties object in carbonEnv is not same as
- // in carbonTable object, so update from carbon env itself.
+ val tableProperties = if (carbonTable.isHivePartitionTable || isComplexTypeProjection) {
+ // in case of partition table and complex type projection, TableProperties object in
+ // carbonEnv is not same as in carbonTable object, so update from carbon env itself.
CarbonEnv.getCarbonTable(Some(carbonTable.getDatabaseName), carbonTable.getTableName)(
sparkSession).getTableInfo
.getFactTable
diff --git a/integration/spark/src/test/resources/secindex/array2.csv b/integration/spark/src/test/resources/secindex/array2.csv
new file mode 100644
index 0000000..e453846
--- /dev/null
+++ b/integration/spark/src/test/resources/secindex/array2.csv
@@ -0,0 +1,4 @@
+1,abc,china$india$us,hello$world
+2,xyz,sri$can,iron$man$jarvis
+3,mno,rus$china,ex$ex2
+4,lok,hk$bang,ex$ex3
diff --git a/integration/spark/src/test/scala/org/apache/carbondata/spark/testsuite/alterTable/TestAlterTableAddColumns.scala b/integration/spark/src/test/scala/org/apache/carbondata/spark/testsuite/alterTable/TestAlterTableAddColumns.scala
index b4504e1..48d9bed 100644
--- a/integration/spark/src/test/scala/org/apache/carbondata/spark/testsuite/alterTable/TestAlterTableAddColumns.scala
+++ b/integration/spark/src/test/scala/org/apache/carbondata/spark/testsuite/alterTable/TestAlterTableAddColumns.scala
@@ -214,6 +214,24 @@ class TestAlterTableAddColumns extends QueryTest with BeforeAndAfterAll {
sql("DROP TABLE IF EXISTS alter_com")
}
+ test("Test alter add complex type and compaction") {
+ sql("DROP TABLE IF EXISTS alter_com")
+ sql("create table alter_com (a int, b string, arr1 array<string>) stored as carbondata")
+ sql("insert into alter_com select 1,'a',array('hi')")
+ sql("insert into alter_com select 2,'b',array('hello','world')")
+ sql("ALTER TABLE alter_com ADD COLUMNS(struct1 STRUCT<a:int, b:string>)")
+ sql("insert into alter_com select 3,'c',array('hi'),null")
+ sql("insert into alter_com select 4,'d',array('hi'),named_struct('s1',4,'s2','d')")
+ sql("alter table alter_com compact 'minor'")
+ checkAnswer(sql("""Select count(*) from alter_com"""), Seq(Row(4)))
+ checkAnswer(sql("Select * from alter_com"),
+ Seq(Row(1, "a", mutable.WrappedArray.make(Array("hi")), null),
+ Row(2, "b", mutable.WrappedArray.make(Array("hello", "world")), null),
+ Row(3, "c", mutable.WrappedArray.make(Array("hi")), null),
+ Row(4, "d", mutable.WrappedArray.make(Array("hi")), Row(4, "d"))))
+ sql("DROP TABLE IF EXISTS alter_com")
+ }
+
def insertIntoTableForArrayType(): Unit = {
sql("insert into alter_com values(4,array(2),array(1,2,3,4),array('abc','def'))")
sql("insert into alter_com values(5,array(1,2),array(1), array('Hulk','Thor'))")
diff --git a/processing/src/main/java/org/apache/carbondata/processing/datatypes/PrimitiveDataType.java b/processing/src/main/java/org/apache/carbondata/processing/datatypes/PrimitiveDataType.java
index 1a09a5d..b0a4263 100644
--- a/processing/src/main/java/org/apache/carbondata/processing/datatypes/PrimitiveDataType.java
+++ b/processing/src/main/java/org/apache/carbondata/processing/datatypes/PrimitiveDataType.java
@@ -414,22 +414,7 @@ public class PrimitiveDataType implements GenericDataType<Object> {
private void updateNullValue(DataOutputStream dataOutputStream, BadRecordLogHolder logHolder)
throws IOException {
- if (this.carbonDimension.getDataType() == DataTypes.STRING
- || this.carbonDimension.getDataType() == DataTypes.VARCHAR) {
- if (DataTypeUtil.isByteArrayComplexChildColumn(dataType)) {
- dataOutputStream.writeInt(CarbonCommonConstants.MEMBER_DEFAULT_VAL_ARRAY.length);
- } else {
- dataOutputStream.writeShort(CarbonCommonConstants.MEMBER_DEFAULT_VAL_ARRAY.length);
- }
- dataOutputStream.write(CarbonCommonConstants.MEMBER_DEFAULT_VAL_ARRAY);
- } else {
- if (DataTypeUtil.isByteArrayComplexChildColumn(dataType)) {
- dataOutputStream.writeInt(CarbonCommonConstants.EMPTY_BYTE_ARRAY.length);
- } else {
- dataOutputStream.writeShort(CarbonCommonConstants.EMPTY_BYTE_ARRAY.length);
- }
- dataOutputStream.write(CarbonCommonConstants.EMPTY_BYTE_ARRAY);
- }
+ CarbonUtil.updateNullValueBasedOnDatatype(dataOutputStream, this.carbonDimension.getDataType());
String message = logHolder.getColumnMessageMap().get(carbonDimension.getColName());
if (null == message) {
message = CarbonDataProcessorUtil