This is an automated email from the ASF dual-hosted git repository.
voonhous pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git
The following commit(s) were added to refs/heads/master by this push:
new 1593524e3244 fix(hadoop-mr): ignore blank Hive projection ids and
report both lists on mismatch (#19463)
1593524e3244 is described below
commit 1593524e3244a7718fe5af1d75a19c703f71933a
Author: Ranga Reddy <[email protected]>
AuthorDate: Sun Aug 30 19:59:47 2026 +0530
fix(hadoop-mr): ignore blank Hive projection ids and report both lists on
mismatch (#19463)
Relates to #14673 (HUDI-1286).
For SELECT COUNT(*) on Hive before 3.0.0 the read-column ids arrive
empty and Hive combines them with Hudi's required projection ids into
e.g. ",2,0,3" (HIVE-22438). Every consumer parses those with
Integer.parseInt, so a blank entry failed with a bare
NumberFormatException carrying none of the projection lists.
cleanProjectionColumnIds is the single point all of them read from, so
the filter goes there. It stripped one leading comma, which leaves
",,2,0" and cannot reach an interior blank at all: Hive prepends ids
while appending names, so an id prepended after an empty one gives
"3,,2,0". It now trims and drops every blank entry, writes the joined
value back only when it changed, and no longer NPEs when the key is
unset. The read-modify-write is synchronized on the conf, the way
addProjectionToJobConf guards its own writes; f41539a9cb5f (#3630)
moved this call out of that latch without restoring a lock.
HoodieParquetInputFormat#getRecordReader now cleans the conf as well.
That is where the bootstrap reader and the parquet schema-evolution
reader are built, and neither goes through a realtime input format, so
on COW those consumers were still reached with the blank in place.
SchemaEvolutionContext read the same key with no default and no filter
in setColumnNameList and setColumnTypeList, so an unset key was an NPE
and a blank id a bare NumberFormatException, in methods that already
report a size mismatch. Both now parse the non-blank tokens and report
that mismatch instead, carrying the counts actually compared rather
than only the raw conf value. getRequireColumn, which runs just before
them, gets the same missing-default fix.
orderFields keeps a filter of its own as defence in depth, for callers
that assemble the csv without going through the conf. Its mismatch
message compared de-duplicated counts while printing the raw name
count, so duplicate names produced two equal numbers; the counts are
now named #distinctFieldNames and #distinctFieldPositions and both
projection lists are included.
Not addressed here and filed as #19506: names and ids are de-duplicated
independently and then paired positionally, which is unsound because
Hive prepends ids while appending names. The fullColNamelist.get(id)
IOOBE in setColumnNameList is out of scope for the same reason.
The mechanism behind #14673 is unconfirmed. An earlier revision
asserted CombineHiveInputFormat accumulating projection entries across
splits, which the issue does not support: the reported frame is
HiveInputFormat.getRecordReader, the base class, and addProjectionField
always sets both keys together. HIVE-22438 and HUDI-313 are what this
repo documents for that query.
Tests: new coverage for orderFields, cleanProjectionColumnIds and
SchemaEvolutionContext, none of which had any before; a blank-id arm on
TestHoodieRealtimeRecordReader#testIncrementalWithReplace driving
cleanProjectionColumnIds and orderFields together; and blank ids in
TestHiveTableSchemaEvolution, the only test that reaches
HoodieParquetInputFormat#getRecordReader on the COW schema-evolution
path. The bootstrap arm remains uncovered, since no test constructs a
BootstrapBaseFileSplit.
---------
Co-authored-by: voon <[email protected]>
---
.../hudi/hadoop/HoodieParquetInputFormat.java | 5 +
.../apache/hudi/hadoop/SchemaEvolutionContext.java | 34 ++++-
.../utils/HoodieRealtimeInputFormatUtils.java | 47 +++++--
.../utils/HoodieRealtimeRecordReaderUtils.java | 16 ++-
.../hudi/hadoop/TestSchemaEvolutionContext.java | 94 +++++++++++++
.../realtime/TestHoodieRealtimeRecordReader.java | 19 ++-
.../utils/TestHoodieRealtimeInputFormatUtils.java | 59 +++++++++
.../utils/TestHoodieRealtimeRecordReaderUtils.java | 146 +++++++++++++++++++++
.../functional/TestHiveTableSchemaEvolution.java | 4 +-
9 files changed, 404 insertions(+), 20 deletions(-)
diff --git
a/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/HoodieParquetInputFormat.java
b/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/HoodieParquetInputFormat.java
index c8089ce390e1..5c6d7451a9e2 100644
---
a/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/HoodieParquetInputFormat.java
+++
b/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/HoodieParquetInputFormat.java
@@ -148,6 +148,11 @@ public class HoodieParquetInputFormat extends
HoodieParquetInputFormatBase {
public RecordReader<NullWritable, ArrayWritable> getRecordReader(final
InputSplit split, final JobConf job,
final
Reporter reporter) throws IOException {
HoodieRealtimeInputFormatUtils.addProjectionField(job,
job.get(hive_metastoreConstants.META_TABLE_PARTITION_COLUMNS, "").split("/"));
+ // The bootstrap and schema-evolution paths below parse the read-column
ids with Integer#parseInt, so the
+ // blank ids HIVE-22438 leaves in the conf have to be dropped here too;
neither path goes through a
+ // realtime input format. The realtime formats clean the conf before
delegating here, so on that path
+ // it is this call that is the no-op.
+ HoodieRealtimeInputFormatUtils.cleanProjectionColumnIds(job);
if (shouldUseFilegroupReader(job, split)) {
try {
if (!(split instanceof FileSplit) || !checkIfHudiTable(split, job)) {
diff --git
a/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/SchemaEvolutionContext.java
b/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/SchemaEvolutionContext.java
index e727965587a7..d3a4532ba3c0 100644
---
a/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/SchemaEvolutionContext.java
+++
b/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/SchemaEvolutionContext.java
@@ -254,13 +254,35 @@ public class SchemaEvolutionContext {
}
}
+ /**
+ * Reads {@code hive.io.file.readcolumn.ids} as its non-blank, trimmed
tokens. The key is unset until
+ * something projects a column, and {@code
HoodieRealtimeInputFormatUtils#cleanProjectionColumnIds} drops
+ * the blank ids HIVE-22438 leaves in it before any reader runs; this keeps
both callers below safe if
+ * either ever reaches them anyway.
+ */
+ private static List<String> parseReadColumnIds(JobConf job) {
+ return
Arrays.stream(job.get(ColumnProjectionUtils.READ_COLUMN_IDS_CONF_STR,
"").split(","))
+ .map(String::trim)
+ .filter(id -> !id.isEmpty())
+ .collect(Collectors.toList());
+ }
+
public void setColumnTypeList(JobConf job, List<Types.Field> fields) {
List<TypeInfo> fullTypeInfos =
TypeInfoUtils.getTypeInfosFromTypeString(job.get(serdeConstants.LIST_COLUMN_TYPES));
- List<Integer> tmpColIdList =
Arrays.stream(job.get(ColumnProjectionUtils.READ_COLUMN_IDS_CONF_STR).split(","))
+ // Blank ids are dropped rather than parsed: HIVE-22438 puts them in this
conf value, and the key is
+ // unset until something projects a column. Either way the size check
below reports the id list instead
+ // of failing with a bare NumberFormatException or an NPE.
+ List<Integer> tmpColIdList = parseReadColumnIds(job).stream()
.map(Integer::parseInt).collect(Collectors.toList());
if (tmpColIdList.size() != fields.size()) {
- throw new HoodieException(String.format("The size of
hive.io.file.readcolumn.ids: %s is not equal to projection columns: %s",
- job.get(ColumnProjectionUtils.READ_COLUMN_IDS_CONF_STR),
fields.stream().map(Types.Field::name).collect(Collectors.joining(","))));
+ // Report the counts that were compared, not just the raw conf value:
the ids are counted after blanks
+ // are dropped, so the string printed here can hold more entries than
the number the check used.
+ throw new HoodieException(String.format(
+ "The size of hive.io.file.readcolumn.ids: %s is not equal to
projection columns: %s. "
+ + "#nonBlankIds: %d, #projectionColumns: %d",
+ job.get(ColumnProjectionUtils.READ_COLUMN_IDS_CONF_STR, ""),
+
fields.stream().map(Types.Field::name).collect(Collectors.joining(",")),
+ tmpColIdList.size(), fields.size()));
}
List<TypeInfo> fieldTypes = new ArrayList<>();
for (int i = 0; i < tmpColIdList.size(); i++) {
@@ -389,7 +411,7 @@ public class SchemaEvolutionContext {
if (fields == null) {
return;
}
- List<String> tmpColIdList =
Arrays.asList(job.get(ColumnProjectionUtils.READ_COLUMN_IDS_CONF_STR).split(","));
+ List<String> tmpColIdList = parseReadColumnIds(job);
if (fields.size() != tmpColIdList.size()) {
return;
}
@@ -417,11 +439,11 @@ public class SchemaEvolutionContext {
public static List<String> getRequireColumn(JobConf jobConf) {
String originColumnString =
jobConf.get(HIVE_TMP_READ_COLUMN_NAMES_CONF_STR);
if (StringUtils.isNullOrEmpty(originColumnString)) {
- jobConf.set(HIVE_TMP_READ_COLUMN_NAMES_CONF_STR,
jobConf.get(ColumnProjectionUtils.READ_COLUMN_NAMES_CONF_STR));
+ jobConf.set(HIVE_TMP_READ_COLUMN_NAMES_CONF_STR,
jobConf.get(ColumnProjectionUtils.READ_COLUMN_NAMES_CONF_STR, ""));
}
String hoodieFullColumnString = jobConf.get(HIVE_TMP_COLUMNS);
if (StringUtils.isNullOrEmpty(hoodieFullColumnString)) {
- jobConf.set(HIVE_TMP_COLUMNS, jobConf.get(serdeConstants.LIST_COLUMNS));
+ jobConf.set(HIVE_TMP_COLUMNS, jobConf.get(serdeConstants.LIST_COLUMNS,
""));
}
String tableColumnString =
jobConf.get(HIVE_TMP_READ_COLUMN_NAMES_CONF_STR);
List<String> tableColumns = Arrays.asList(tableColumnString.split(","));
diff --git
a/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/utils/HoodieRealtimeInputFormatUtils.java
b/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/utils/HoodieRealtimeInputFormatUtils.java
index f40eca130872..791aba0d0400 100644
---
a/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/utils/HoodieRealtimeInputFormatUtils.java
+++
b/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/utils/HoodieRealtimeInputFormatUtils.java
@@ -128,16 +128,47 @@ public class HoodieRealtimeInputFormatUtils extends
HoodieInputFormatUtils {
}
/**
- * Hive will append read columns' ids to old columns' ids during
getRecordReader. In some cases, e.g. SELECT COUNT(*),
- * the read columns' id is an empty string and Hive will combine it with
Hoodie required projection ids and becomes
- * e.g. ",2,0,3" and will cause an error. Actually this method is a
temporary solution because the real bug is from
- * Hive. Hive has fixed this bug after 3.0.0, but the version before that
would still face this problem. (HIVE-22438)
+ * Drops blank entries from the read-column id list held in {@code conf}.
+ *
+ * <p>For {@code SELECT COUNT(*)} on Hive before 3.0.0 the read-column ids
arrive empty and Hive combines
+ * them into e.g. {@code ",2,0,3"} (HIVE-22438). Every consumer parses those
ids with
+ * {@code Integer#parseInt}, so a blank entry fails with a bare {@code
NumberFormatException} that carries
+ * none of the projection lists:
+ *
+ * <ul>
+ * <li>{@code SchemaEvolutionContext#setColumnNameList} and {@code
#setColumnTypeList}, both reached from
+ * {@code doEvolutionForParquetFormat}</li>
+ * <li>{@code HoodieColumnProjectionUtils#getReadColumnIDs}, on the
bootstrap path</li>
+ * <li>{@code HoodieRealtimeRecordReaderUtils#orderFields}, the path in
the reported issue</li>
+ * </ul>
+ *
+ * <p>Each of those is reached from a {@code getRecordReader} that cleans
the conf first:
+ * {@code HoodieParquetInputFormat} for the parquet and bootstrap paths, and
+ * {@code HoodieParquetRealtimeInputFormat} / {@code
HoodieHFileRealtimeInputFormat} for the realtime ones.
+ * {@code HoodieCombineHiveInputFormat} builds its per-split readers through
those same formats and hands
+ * them the conf they cleaned, so it is covered as well.
+ *
+ * <p>This is a workaround: the underlying bug is in Hive, fixed after
3.0.0, but earlier versions still
+ * hit it. Stripping a single leading comma is not enough. Hive prepends ids
while appending names, so repeated
+ * empty appends give {@code ",,2,0"}, and an id prepended after an empty
one gives {@code "3,,2,0"} where
+ * the blank is interior and no amount of leading-comma stripping reaches it.
+ *
+ * <p>Hive on Spark calls {@code getRecordReader} from several threads
sharing one {@code JobConf}, so the
+ * read-modify-write below is synchronized on the conf the same way {@code
addProjectionToJobConf} guards
+ * its own writes. The write-back is skipped when nothing changed, so an
unset key stays unset rather than
+ * being written back as empty.
*/
public static void cleanProjectionColumnIds(Configuration conf) {
- String columnIds =
conf.get(ColumnProjectionUtils.READ_COLUMN_IDS_CONF_STR);
- if (!columnIds.isEmpty() && columnIds.charAt(0) == ',') {
- conf.set(ColumnProjectionUtils.READ_COLUMN_IDS_CONF_STR,
columnIds.substring(1));
- LOG.debug("The projection Ids: {{}} start with ','. First comma is
removed", columnIds);
+ synchronized (conf) {
+ String columnIds =
conf.get(ColumnProjectionUtils.READ_COLUMN_IDS_CONF_STR, "");
+ String cleanedColumnIds = Arrays.stream(columnIds.split(","))
+ .map(String::trim)
+ .filter(id -> !id.isEmpty())
+ .collect(Collectors.joining(","));
+ if (!cleanedColumnIds.equals(columnIds)) {
+ conf.set(ColumnProjectionUtils.READ_COLUMN_IDS_CONF_STR,
cleanedColumnIds);
+ LOG.debug("The projection Ids: {{}} contained blank entries. Cleaned
to: {{}}", columnIds, cleanedColumnIds);
+ }
}
}
}
diff --git
a/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/utils/HoodieRealtimeRecordReaderUtils.java
b/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/utils/HoodieRealtimeRecordReaderUtils.java
index 59dd14cff733..c3f643abcda6 100644
---
a/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/utils/HoodieRealtimeRecordReaderUtils.java
+++
b/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/utils/HoodieRealtimeRecordReaderUtils.java
@@ -273,15 +273,25 @@ public class HoodieRealtimeRecordReaderUtils {
// /org/apache/hadoop/hive/serde2/ColumnProjectionUtils.java#L188}
// Field Names -> {@link
https://github.com/apache/hive/blob/f37c5de6c32b9395d1b34fa3c02ed06d1bfbf6eb/serde/src/java
// /org/apache/hadoop/hive/serde2/ColumnProjectionUtils.java#L229}
- String[] fieldOrdersWithDups = fieldOrderCsv.isEmpty() ? new String[0] :
fieldOrderCsv.split(",");
+ // Defence in depth.
HoodieRealtimeInputFormatUtils#cleanProjectionColumnIds now drops blank ids
from the
+ // JobConf before any reader runs, which is what HIVE-22438 produces for
SELECT COUNT(*) on Hive before
+ // 3.0.0, so blanks should no longer arrive here. Callers that assemble
the csv without going through that
+ // conf still can, and a blank token would otherwise reach
Integer.parseInt below and fail with a bare
+ // NumberFormatException carrying neither projection list. Trim before
filtering so a padded id parses and
+ // de-duplicates as the same entry rather than as a distinct one.
+ String[] fieldOrdersWithDups = Arrays.stream(fieldOrderCsv.split(","))
+ .map(String::trim).filter(id -> !id.isEmpty()).toArray(String[]::new);
Set<String> fieldOrdersSet = new
LinkedHashSet<>(Arrays.asList(fieldOrdersWithDups));
String[] fieldOrders = fieldOrdersSet.toArray(new String[0]);
List<String> fieldNames = fieldNameCsv.isEmpty() ? new ArrayList<>() :
Arrays.stream(fieldNameCsv.split(",")).collect(Collectors.toList());
Set<String> fieldNamesSet = new LinkedHashSet<>(fieldNames);
if (fieldNamesSet.size() != fieldOrders.length) {
+ // Report the de-duplicated counts, since those are what was compared:
the raw name count can print
+ // two equal numbers for a real mismatch.
throw new HoodieException(String
- .format("Error ordering fields for storage read. #fieldNames: %d,
#fieldPositions: %d",
- fieldNames.size(), fieldOrders.length));
+ .format("Error ordering fields for storage read.
#distinctFieldNames: %d, #distinctFieldPositions: %d, "
+ + "read column names: [%s], read column ids: [%s]",
+ fieldNamesSet.size(), fieldOrders.length, fieldNameCsv,
fieldOrderCsv));
}
TreeMap<Integer, String> orderedFieldMap = new TreeMap<>();
String[] fieldNamesArray = fieldNamesSet.toArray(new String[0]);
diff --git
a/hudi-hadoop-mr/src/test/java/org/apache/hudi/hadoop/TestSchemaEvolutionContext.java
b/hudi-hadoop-mr/src/test/java/org/apache/hudi/hadoop/TestSchemaEvolutionContext.java
new file mode 100644
index 000000000000..2b0122478eaf
--- /dev/null
+++
b/hudi-hadoop-mr/src/test/java/org/apache/hudi/hadoop/TestSchemaEvolutionContext.java
@@ -0,0 +1,94 @@
+/*
+ * 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.hudi.hadoop;
+
+import org.apache.hudi.common.schema.internal.Types;
+import org.apache.hudi.exception.HoodieException;
+
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hive.serde.serdeConstants;
+import org.apache.hadoop.hive.serde2.ColumnProjectionUtils;
+import org.apache.hadoop.mapred.FileSplit;
+import org.apache.hadoop.mapred.JobConf;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.List;
+
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Covers how {@link SchemaEvolutionContext#setColumnTypeList} reads {@code
hive.io.file.readcolumn.ids}.
+ * Every id there is parsed with {@code Integer#parseInt}, so the blank
entries HIVE-22438 leaves behind and
+ * the unset key both used to surface as a bare {@code NumberFormatException}
or an NPE rather than as the
+ * size mismatch the method already reports.
+ */
+public class TestSchemaEvolutionContext {
+
+ // col2 is a record whose nested field was renamed, so setColumnTypeList has
something to write back:
+ // primitive types are returned unchanged, and an unchanged rewrite cannot
tell a right pairing from a wrong one.
+ private static final List<Types.Field> TWO_FIELDS = Arrays.asList(
+ Types.Field.get(0, "col1", Types.StringType.get()),
+ Types.Field.get(1, "col2", Types.RecordType.get(
+ Types.Field.get(2, "renamed", Types.StringType.get()))));
+
+ private JobConf job;
+ private SchemaEvolutionContext context;
+
+ @BeforeEach
+ public void setUp() throws IOException {
+ job = new JobConf();
+ // Keeps the constructor off the table: it is the projection-id parsing
below that is under test.
+ job.setBoolean("hudi.hive.schema.evolution", false);
+ job.set(serdeConstants.LIST_COLUMN_TYPES,
"string,struct<original:string>");
+ context = new SchemaEvolutionContext(new FileSplit(new
Path("file:///tmp/unused"), 0, 0, (String[]) null), job);
+ }
+
+ @Test
+ public void testSetColumnTypeListWithUnsetReadColumnIds() {
+ job.unset(ColumnProjectionUtils.READ_COLUMN_IDS_CONF_STR);
+ HoodieException thrown = assertThrows(HoodieException.class, () ->
context.setColumnTypeList(job, TWO_FIELDS));
+ assertTrue(thrown.getMessage().contains("is not equal to projection
columns"),
+ () -> "Expected the size mismatch rather than an NPE, got: " +
thrown.getMessage());
+ }
+
+ @Test
+ public void testSetColumnTypeListWithBlankReadColumnId() {
+ job.set(ColumnProjectionUtils.READ_COLUMN_IDS_CONF_STR, ",0");
+ HoodieException thrown = assertThrows(HoodieException.class, () ->
context.setColumnTypeList(job, TWO_FIELDS));
+ assertTrue(thrown.getMessage().contains("is not equal to projection
columns"),
+ () -> "Expected the size mismatch rather than a NumberFormatException,
got: " + thrown.getMessage());
+ assertTrue(thrown.getMessage().contains("#nonBlankIds: 1,
#projectionColumns: 2"),
+ () -> "the message should carry the counts that were compared, got: "
+ thrown.getMessage());
+ }
+
+ @Test
+ public void testSetColumnTypeListIgnoresBlankAndPaddedReadColumnIds() {
+ job.set(ColumnProjectionUtils.READ_COLUMN_IDS_CONF_STR, ", 0 ,1");
+ assertDoesNotThrow(() -> context.setColumnTypeList(job, TWO_FIELDS));
+ assertEquals("string,struct<renamed:string>",
job.get(serdeConstants.LIST_COLUMN_TYPES),
+ "id 1 should still pair with col2, so the renamed nested field lands
in the second slot");
+ }
+}
diff --git
a/hudi-hadoop-mr/src/test/java/org/apache/hudi/hadoop/realtime/TestHoodieRealtimeRecordReader.java
b/hudi-hadoop-mr/src/test/java/org/apache/hudi/hadoop/realtime/TestHoodieRealtimeRecordReader.java
index 4b1417700178..15cf2472e2bc 100644
---
a/hudi-hadoop-mr/src/test/java/org/apache/hudi/hadoop/realtime/TestHoodieRealtimeRecordReader.java
+++
b/hudi-hadoop-mr/src/test/java/org/apache/hudi/hadoop/realtime/TestHoodieRealtimeRecordReader.java
@@ -86,6 +86,7 @@ import org.junit.jupiter.api.io.TempDir;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.Arguments;
import org.junit.jupiter.params.provider.MethodSource;
+import org.junit.jupiter.params.provider.ValueSource;
import java.io.File;
import java.io.FileOutputStream;
@@ -750,8 +751,15 @@ public class TestHoodieRealtimeRecordReader {
}
}
- @Test
- public void testIncrementalWithReplace() throws Exception {
+ /**
+ * The {@code blankProjectionIds} arm drives the two changed methods
together: Hive's read-column ids reach
+ * {@code HoodieParquetRealtimeInputFormat#getRecordReader} with the blanks
HIVE-22438 leaves behind,
+ * {@code cleanProjectionColumnIds} drops them, and {@code orderFields}
pairs what is left. Two blanks, not
+ * one, because the previous sanitiser stripped a single leading comma.
+ */
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void testIncrementalWithReplace(boolean blankProjectionIds) throws
Exception {
// initial commit
HoodieSchema schema =
HoodieSchemaUtils.addMetadataFields(SchemaTestUtil.getEvolvedSchema());
HoodieTestUtils.init(storageConf, basePath.toString(),
HoodieTableType.MERGE_ON_READ);
@@ -779,12 +787,17 @@ public class TestHoodieRealtimeRecordReader {
JobConf newJobConf = new JobConf(baseJobConf);
List<HoodieSchemaField> fields = schema.getFields();
setHiveColumnNameProps(fields, newJobConf, false);
+ if (blankProjectionIds) {
+ newJobConf.set(ColumnProjectionUtils.READ_COLUMN_IDS_CONF_STR,
+ ",," +
newJobConf.get(ColumnProjectionUtils.READ_COLUMN_IDS_CONF_STR));
+ }
newJobConf.set("columns.types",
"string,string,string,string,string,string,string,string,bigint,string,string");
RecordReader<NullWritable, ArrayWritable> reader =
inputFormat.getRecordReader(splits[0], newJobConf, Reporter.NULL);
// use reader to read log file.
NullWritable key = reader.createKey();
ArrayWritable value = reader.createValue();
+ int recordCnt = 0;
while (reader.next(key, value)) {
Writable[] values = value.get();
// since we set incremental start commit as 0 and commit_number as 1.
@@ -792,7 +805,9 @@ public class TestHoodieRealtimeRecordReader {
assertEquals("100", values[0].toString());
key = reader.createKey();
value = reader.createValue();
+ recordCnt++;
}
+ assertEquals(100, recordCnt);
reader.close();
}
diff --git
a/hudi-hadoop-mr/src/test/java/org/apache/hudi/hadoop/utils/TestHoodieRealtimeInputFormatUtils.java
b/hudi-hadoop-mr/src/test/java/org/apache/hudi/hadoop/utils/TestHoodieRealtimeInputFormatUtils.java
index deecaca5c706..350718c4fea1 100644
---
a/hudi-hadoop-mr/src/test/java/org/apache/hudi/hadoop/utils/TestHoodieRealtimeInputFormatUtils.java
+++
b/hudi-hadoop-mr/src/test/java/org/apache/hudi/hadoop/utils/TestHoodieRealtimeInputFormatUtils.java
@@ -22,9 +22,19 @@ import org.apache.hudi.common.testutils.HoodieTestUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hive.metastore.api.hive_metastoreConstants;
+import org.apache.hadoop.hive.serde2.ColumnProjectionUtils;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
+
+import java.util.stream.Stream;
+
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
public class TestHoodieRealtimeInputFormatUtils {
@@ -45,4 +55,53 @@ public class TestHoodieRealtimeInputFormatUtils {
hadoopConf.set(hive_metastoreConstants.META_TABLE_PARTITION_COLUMNS, "");
HoodieRealtimeInputFormatUtils.addProjectionField(hadoopConf,
hadoopConf.get(hive_metastoreConstants.META_TABLE_PARTITION_COLUMNS,
"").split("/"));
}
+
+ private String clean(String columnIds) {
+ hadoopConf.set(ColumnProjectionUtils.READ_COLUMN_IDS_CONF_STR, columnIds);
+ HoodieRealtimeInputFormatUtils.cleanProjectionColumnIds(hadoopConf);
+ return hadoopConf.get(ColumnProjectionUtils.READ_COLUMN_IDS_CONF_STR);
+ }
+
+ private static Stream<Arguments> projectionIdCases() {
+ return Stream.of(
+ Arguments.of(",2,0", "2,0", "a leading blank id should be dropped"),
+ Arguments.of(",,2,0", "2,0",
+ "Hive appending empty ids repeatedly yields more than one leading
blank"),
+ Arguments.of("3,,2,0", "3,2,0",
+ "an interior blank, which leading-comma stripping never reached;
the resulting pairing is still "
+ + "unsound per #19506, this only stops the bare
NumberFormatException"),
+ Arguments.of(" 2 , 0 ", "2,0",
+ "ids are trimmed, so a padded id parses and de-duplicates as the
same entry"),
+ Arguments.of("2,0", "2,0", "a list with nothing to drop keeps its
value"),
+ Arguments.of("", "", "an empty list keeps its value"));
+ }
+
+ /**
+ * HIVE-22438: for {@code SELECT COUNT(*)} on Hive before 3.0.0 the
read-column ids arrive empty and Hive
+ * combines them into e.g. {@code ",2,0,3"}. Every consumer of this conf
value parses the ids with
+ * {@code Integer#parseInt}, so any blank entry left behind fails with a
bare {@code NumberFormatException}.
+ *
+ * <p>One case per shape, so a regression in the first does not hide the
rest: only three of the six differ
+ * from what the previous single-leading-comma sanitiser produced.
+ */
+ @ParameterizedTest(name = "[{index}] \"{0}\" -> \"{1}\"")
+ @MethodSource("projectionIdCases")
+ public void testCleanProjectionColumnIdsDropsBlankEntries(String columnIds,
String expected, String why) {
+ assertEquals(expected, clean(columnIds), why);
+ }
+
+ /**
+ * The key is unset until something projects a column, and {@code conf.get}
had no default here, so this
+ * threw {@code NullPointerException} before. Same shape as the fix applied
to {@code addProjectionField}.
+ *
+ * <p>This also pins the write-back guard: without it the empty default
would be written back and the key
+ * would no longer read as unset.
+ */
+ @Test
+ public void testCleanProjectionColumnIdsWithUnsetKey() {
+ hadoopConf.unset(ColumnProjectionUtils.READ_COLUMN_IDS_CONF_STR);
+ assertDoesNotThrow(() ->
HoodieRealtimeInputFormatUtils.cleanProjectionColumnIds(hadoopConf));
+ assertNull(hadoopConf.get(ColumnProjectionUtils.READ_COLUMN_IDS_CONF_STR),
+ "an unset key should stay unset rather than being written back as
empty");
+ }
}
diff --git
a/hudi-hadoop-mr/src/test/java/org/apache/hudi/hadoop/utils/TestHoodieRealtimeRecordReaderUtils.java
b/hudi-hadoop-mr/src/test/java/org/apache/hudi/hadoop/utils/TestHoodieRealtimeRecordReaderUtils.java
new file mode 100644
index 000000000000..a81a2d49f9bb
--- /dev/null
+++
b/hudi-hadoop-mr/src/test/java/org/apache/hudi/hadoop/utils/TestHoodieRealtimeRecordReaderUtils.java
@@ -0,0 +1,146 @@
+/*
+ * 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.hudi.hadoop.utils;
+
+import org.apache.hudi.exception.HoodieException;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.Arrays;
+import java.util.Collections;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Tests {@link HoodieRealtimeRecordReaderUtils#orderFields}, which maps Hive's
+ * {@code hive.io.file.readcolumn.names} and {@code
hive.io.file.readcolumn.ids} onto an ordered
+ * projection list.
+ */
+public class TestHoodieRealtimeRecordReaderUtils {
+
+ @Test
+ public void testOrderFieldsSortsNamesByTheirHivePosition() {
+ assertEquals(Arrays.asList("rider", "driver", "fare"),
+ HoodieRealtimeRecordReaderUtils.orderFields("driver,fare,rider",
"1,2,0", Collections.emptyList()));
+ }
+
+ @Test
+ public void testOrderFieldsReturnsEmptyForEmptyInput() {
+ assertEquals(Collections.emptyList(),
+ HoodieRealtimeRecordReaderUtils.orderFields("", "",
Collections.emptyList()));
+ }
+
+ /**
+ * Hive can repeat a name in the read-column list while keeping ids unique,
which the method
+ * deliberately tolerates by de-duplicating both sides before pairing them.
The duplicate is deliberately
+ * not last: that is the one position where pairing from the de-duplicated
list and from the raw one give
+ * the same answer, so a duplicate on the tail pins the count but not the
pairing.
+ */
+ @Test
+ public void testOrderFieldsDeduplicatesRepeatedNames() {
+ assertEquals(Arrays.asList("rider", "driver"),
+ HoodieRealtimeRecordReaderUtils.orderFields("rider,rider,driver",
"0,1", Collections.emptyList()));
+ }
+
+ /**
+ * The counts compared are the de-duplicated ones, so the failure has to
report those. Reporting the raw
+ * name count instead prints two equal numbers for a real mismatch, which is
unusable when diagnosing
+ * something like HUDI-1286. This is the case that fails without the
production change.
+ */
+ @Test
+ public void testOrderFieldsMismatchReportsDistinctCountsWhenNamesRepeat() {
+ HoodieException thrown = assertThrows(HoodieException.class, () ->
+ HoodieRealtimeRecordReaderUtils.orderFields("rider,driver,fare,fare",
"0,1,2,3", Collections.emptyList()));
+ assertTrue(thrown.getMessage().contains("#distinctFieldNames: 3"),
+ () -> "Expected the de-duplicated name count, got: " +
thrown.getMessage());
+ assertTrue(thrown.getMessage().contains("#distinctFieldPositions: 4"),
+ () -> "Expected the position count, got: " + thrown.getMessage());
+ }
+
+ /** A mismatch with no duplicates on either side still has to carry both
projection lists. */
+ @Test
+ public void testOrderFieldsMismatchReportsBothProjectionLists() {
+ HoodieException thrown = assertThrows(HoodieException.class, () ->
+ HoodieRealtimeRecordReaderUtils.orderFields("rider,driver", "0,1,2",
Collections.emptyList()));
+ assertTrue(thrown.getMessage().contains("read column names:
[rider,driver]")
+ && thrown.getMessage().contains("read column ids: [0,1,2]"),
+ () -> "Expected both projection lists, got: " + thrown.getMessage());
+ }
+
+ /**
+ * HIVE-22438: for {@code SELECT COUNT(*)} on Hive before 3.0.0 the
read-column ids arrive empty and Hive
+ * combines them into e.g. {@code ",,2,0,3,5"}. {@code
cleanProjectionColumnIds} strips every blank id from
+ * the conf, so no production path delivers a blank here any more. This is
defence in depth for callers
+ * that build the csv themselves.
+ */
+ @Test
+ public void testOrderFieldsIgnoresBlankIdTokens() {
+ assertEquals(Arrays.asList("c", "b"),
+ HoodieRealtimeRecordReaderUtils.orderFields("b,c", ",2,0",
Collections.emptyList()),
+ "a leading blank id token should be ignored rather than parsed");
+ assertEquals(Arrays.asList("c", "b"),
+ HoodieRealtimeRecordReaderUtils.orderFields("b,c", ",,2,0",
Collections.emptyList()),
+ "more than one blank token can arrive when Hive appends empty ids
repeatedly");
+ assertEquals(Arrays.asList("b", "a", "c", "ts"),
+ HoodieRealtimeRecordReaderUtils.orderFields("a,b,c,ts", ",2,0,3,5",
Collections.emptyList()),
+ "the numbers reported in #14673: four names against five id tokens,
one of them blank");
+ assertEquals(Arrays.asList("c", "b"),
+ HoodieRealtimeRecordReaderUtils.orderFields("b,c", " 2 , 0 ",
Collections.emptyList()),
+ "ids are trimmed, so a padded id parses and de-duplicates as the same
entry");
+ }
+
+ /**
+ * A blank token only reaches {@code Integer.parseInt} when it makes the two
counts line up, which needs one
+ * more name than the cases above. That is the shape that failed with a bare
{@code NumberFormatException}
+ * carrying neither projection list; dropping the blank turns it into the
{@code HoodieException} that does
+ * carry both.
+ */
+ @Test
+ public void testOrderFieldsMismatchAfterBlankIdFilteringReportsRawIdList() {
+ HoodieException thrown = assertThrows(HoodieException.class, () ->
+ HoodieRealtimeRecordReaderUtils.orderFields("a,b,c", ",2,0",
Collections.emptyList()));
+ assertTrue(thrown.getMessage().contains("read column ids: [,2,0]"),
+ () -> "Expected the raw id list, got: " + thrown.getMessage());
+ }
+
+ /**
+ * Hive tolerates duplicate ids as well as duplicate names, which is why
both sides are de-duplicated
+ * before pairing. Only the name side was pinned. As above, the duplicate is
not on the tail.
+ */
+ @Test
+ public void testOrderFieldsDeduplicatesRepeatedIds() {
+ assertEquals(Arrays.asList("rider", "driver"),
+ HoodieRealtimeRecordReaderUtils.orderFields("rider,driver", "0,0,1",
Collections.emptyList()));
+ }
+
+ /**
+ * HUDI-5308 (#7355) removed the filter that dropped partitioning fields
from the name list before the
+ * comparison, so a partition column in that list now counts towards it.
Pins that removal.
+ */
+ @Test
+ public void testOrderFieldsNoLongerFiltersPartitionFields() {
+ HoodieException thrown = assertThrows(HoodieException.class, () ->
HoodieRealtimeRecordReaderUtils.orderFields(
+ "rider,driver,partition_path", "0,1",
Collections.singletonList("partition_path")));
+ assertTrue(thrown.getMessage().contains("#distinctFieldNames: 3,
#distinctFieldPositions: 2"),
+ () -> "the partition column should still count towards the comparison,
got: " + thrown.getMessage());
+ }
+}
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/functional/TestHiveTableSchemaEvolution.java
b/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/functional/TestHiveTableSchemaEvolution.java
index d1b2e73a0812..fafd3c1e32a6 100644
---
a/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/functional/TestHiveTableSchemaEvolution.java
+++
b/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/functional/TestHiveTableSchemaEvolution.java
@@ -160,7 +160,9 @@ public class TestHiveTableSchemaEvolution {
jobConf.set(HoodieCommonConfig.SCHEMA_EVOLUTION_ENABLE.key(), "true");
jobConf.set(ColumnProjectionUtils.READ_ALL_COLUMNS, "false");
jobConf.set(ColumnProjectionUtils.READ_COLUMN_NAMES_CONF_STR,
"col1,col2_new");
- jobConf.set(ColumnProjectionUtils.READ_COLUMN_IDS_CONF_STR, "6,7");
+ // Leading blanks are what HIVE-22438 puts in this conf value. The cow arm
builds a HoodieParquetInputFormat
+ // directly, so this is the only test that drives cleanProjectionColumnIds
ahead of SchemaEvolutionContext.
+ jobConf.set(ColumnProjectionUtils.READ_COLUMN_IDS_CONF_STR, ",,6,7");
jobConf.set(serdeConstants.LIST_COLUMNS,
"_hoodie_commit_time,_hoodie_commit_seqno,"
+
"_hoodie_record_key,_hoodie_partition_path,_hoodie_file_name,col0,col1,col2_new");
jobConf.set(serdeConstants.LIST_COLUMN_TYPES,
"string,string,string,string,string,int,double,string");
