Skip to content

Commit 581efe2

Browse files
nsivabalannsivabalan
committed
addressing feedback
1 parent 245cb3f commit 581efe2

4 files changed

Lines changed: 39 additions & 35 deletions

File tree

‎hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/io/storage/row/HoodieRowCreateHandle.java‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -113,9 +113,9 @@ public HoodieRowCreateHandle(HoodieTable table, HoodieWriteConfig writeConfig, S
113113
*/
114114
public void write(InternalRow record) throws IOException {
115115
try {
116-
String partitionPath = record.getUTF8String(3).toString();
117-
String seqId = HoodieRecord.generateSequenceId(instantTime, taskPartitionId, SEQGEN.getAndIncrement());
118-
String recordKey = record.getUTF8String(2).toString();
116+
final String partitionPath = String.valueOf(record.getUTF8String(3));
117+
final String seqId = HoodieRecord.generateSequenceId(instantTime, taskPartitionId, SEQGEN.getAndIncrement());
118+
final String recordKey = String.valueOf(record.getUTF8String(2));
119119
HoodieInternalRow internalRow = new HoodieInternalRow(instantTime, seqId, recordKey, partitionPath, path.getName(),
120120
record);
121121
try {

‎hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/keygen/BuiltinKeyGenerator.java‎

Lines changed: 34 additions & 30 deletions
Original file line numberDiff line numberDiff line change
@@ -36,6 +36,7 @@
3636
import java.util.HashMap;
3737
import java.util.List;
3838
import java.util.Map;
39+
import java.util.concurrent.atomic.AtomicBoolean;
3940

4041
import scala.Function1;
4142

@@ -45,10 +46,12 @@
4546
*/
4647
public abstract class BuiltinKeyGenerator extends BaseKeyGenerator implements SparkKeyGeneratorInterface {
4748

49+
private static final String DOT_STRING = ".";
4850
private static final String STRUCT_NAME = "hoodieRowTopLevelField";
4951
private static final String NAMESPACE = "hoodieRow";
5052
private Function1<Row, GenericRecord> converterFn = null;
5153
protected StructType structType;
54+
private static AtomicBoolean validatePartitionFields = new AtomicBoolean(false);
5255

5356
protected Map<String, Pair<List<Integer>, DataType>> recordKeySchemaInfo = new HashMap<>();
5457
protected Map<String, Pair<List<Integer>, DataType>> partitionPathSchemaInfo = new HashMap<>();
@@ -108,37 +111,40 @@ public String getPartitionPath(InternalRow internalRow, StructType structType) {
108111

109112
void buildFieldSchemaInfoIfNeeded(StructType structType) {
110113
if (this.structType == null) {
111-
// parse simple fields
112-
getRecordKeyFields().stream()
113-
.filter(f -> !(f.contains(".")))
114+
getRecordKeyFields()
115+
.stream().filter(f -> !f.isEmpty())
114116
.forEach(f -> {
115-
if (structType.getFieldIndex(f).isDefined()) {
116-
int fieldIndex = (int) structType.getFieldIndex(f).get();
117-
recordKeySchemaInfo.put(f, Pair.of(Collections.singletonList((fieldIndex)), structType.fields()[fieldIndex].dataType()));
117+
if (f.contains(DOT_STRING)) {
118+
// nested field
119+
recordKeySchemaInfo.put(f, RowKeyGeneratorHelper.getNestedFieldSchemaInfo(structType, f, true));
118120
} else {
119-
throw new HoodieKeyException("recordKey value not found for field: \"" + f + "\"");
121+
// simple field
122+
if (structType.getFieldIndex(f).isDefined()) {
123+
int fieldIndex = (int) structType.getFieldIndex(f).get();
124+
recordKeySchemaInfo.put(f, Pair.of(Collections.singletonList((fieldIndex)), structType.fields()[fieldIndex].dataType()));
125+
} else {
126+
throw new HoodieKeyException("recordKey value not found for field: \"" + f + "\"");
127+
}
120128
}
121129
});
122-
// parse nested fields
123-
getRecordKeyFields().stream()
124-
.filter(f -> f.contains("."))
125-
.forEach(f -> recordKeySchemaInfo.put(f, RowKeyGeneratorHelper.getNestedFieldSchemaInfo(structType, f, true)));
126-
// parse simple fields
127130
if (getPartitionPathFields() != null) {
128-
getPartitionPathFields().stream().filter(f -> !f.isEmpty()).filter(f -> !(f.contains(".")))
131+
getPartitionPathFields().stream().filter(f -> !f.isEmpty())
129132
.forEach(f -> {
130-
if (structType.getFieldIndex(f).isDefined()) {
131-
int fieldIndex = (int) structType.getFieldIndex(f).get();
133+
// nested field
134+
if (f.contains(DOT_STRING)) {
132135
partitionPathSchemaInfo.put(f,
133-
Pair.of(Collections.singletonList(fieldIndex), structType.fields()[fieldIndex].dataType()));
136+
RowKeyGeneratorHelper.getNestedFieldSchemaInfo(structType, f, false));
134137
} else {
135-
partitionPathSchemaInfo.put(f, Pair.of(Collections.singletonList(-1), null));
138+
// simple field
139+
if (structType.getFieldIndex(f).isDefined()) {
140+
int fieldIndex = (int) structType.getFieldIndex(f).get();
141+
partitionPathSchemaInfo.put(f,
142+
Pair.of(Collections.singletonList(fieldIndex), structType.fields()[fieldIndex].dataType()));
143+
} else {
144+
partitionPathSchemaInfo.put(f, Pair.of(Collections.singletonList(-1), null));
145+
}
136146
}
137147
});
138-
// parse nested fields
139-
getPartitionPathFields().stream().filter(f -> !f.isEmpty()).filter(f -> f.contains("."))
140-
.forEach(f -> partitionPathSchemaInfo.put(f,
141-
RowKeyGeneratorHelper.getNestedFieldSchemaInfo(structType, f, false)));
142148
}
143149
this.structType = structType;
144150
}
@@ -152,15 +158,13 @@ protected String getPartitionPathInternal(InternalRow row, StructType structType
152158
}
153159

154160
protected void validatePartitionFieldsForInternalRow() {
155-
partitionPathSchemaInfo.values().forEach(entry -> {
156-
if (entry.getKey().size() > 1) {
157-
throw new IllegalArgumentException("Nested column for partitioning is not supported with disabling meta columns");
158-
}
159-
});
160-
}
161-
162-
void buildFieldDataTypesMapIfNeeded(StructType structType) {
163-
buildFieldSchemaInfoIfNeeded(structType);
161+
if (!validatePartitionFields.getAndSet(true)) {
162+
partitionPathSchemaInfo.values().forEach(entry -> {
163+
if (entry.getKey().size() > 1) {
164+
throw new IllegalArgumentException("Nested column for partitioning is not supported with disabling meta columns");
165+
}
166+
});
167+
}
164168
}
165169
}
166170

‎hudi-common/src/main/java/org/apache/hudi/common/table/HoodieTableMetaClient.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -389,7 +389,7 @@ public void validateTableProperties(Properties properties, WriteOperationType op
389389
// meta fields can be disabled only with SimpleKeyGenerator, NonPartitioned and ComplexKeyGen.
390390
if (!getTableConfig().populateMetaFields()) {
391391
String keyGenClass = properties.getProperty(HoodieTableConfig.KEY_GENERATOR_CLASS_NAME.key(), "org.apache.hudi.keygen.SimpleKeyGenerator");
392-
if (!keyGenClass.equals("org.apache.hudi.keygen.SimpleKeyGenerator") && !keyGenClass.equals("org.apache.hudi.keygen..NonpartitionedKeyGenerator")
392+
if (!keyGenClass.equals("org.apache.hudi.keygen.SimpleKeyGenerator") && !keyGenClass.equals("org.apache.hudi.keygen.NonpartitionedKeyGenerator")
393393
&& !keyGenClass.equals("org.apache.hudi.keygen.ComplexKeyGenerator")) {
394394
throw new HoodieException("Only simple, non partitioned and complex key generator is supported when meta fields are disabled. KeyGenerator used : "
395395
+ properties.getProperty(HoodieTableConfig.KEY_GENERATOR_CLASS_NAME.key()));

‎hudi-spark-datasource/hudi-spark-common/src/main/java/org/apache/hudi/internal/BulkInsertDataInternalWriterHelper.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -122,7 +122,7 @@ public void write(InternalRow record) throws IOException {
122122
try {
123123
String partitionPath = null;
124124
if (populateMetaFields) { // usual path where meta fields are pre populated in prep step.
125-
partitionPath = record.getUTF8String(3).toString();
125+
partitionPath = String.valueOf(record.getUTF8String(3));
126126
} else { // if meta columns are disabled.
127127
if (!keyGeneratorOpt.isPresent()) { // NoPartitionerKeyGen
128128
partitionPath = "";

0 commit comments

Comments
 (0)