wombatu-kun opened a new issue, #20313:
URL: https://github.com/apache/hudi/issues/20313
### Bug Description
**What happened:**
On a Flink table with a bucket index, one primary key and two or more
partition fields, an equality filter on the primary key returns no rows, while
an unfiltered scan returns them. Such a table uses the complex key generator,
whose default `FIELD_PREFIXED` encoding stores the record key as `uuid:id1`.
The bucket writers hash that stored key through
`KeyGenUtils.extractRecordKeysByFields`, which returns it unsplit because it
only splits keys holding both `,` and `:`. `PrimaryKeyPruners.getBucketIdFunc`
hashes the bare filter literal `id1` instead, so bucket pruning keeps another
bucket's file group. With the default 4 buckets, every key of the reproducer
below lands in a different bucket than the one the pruner picks, on both
COPY_ON_WRITE and MERGE_ON_READ. Tables that store bare keys (`VALUE_ONLY`) are
not affected.
**What you expected:**
`SELECT * FROM t1 WHERE uuid = 'id1'` returns the row that `SELECT * FROM
t1` returns.
**Steps to reproduce:**
```sql
CREATE TABLE t1 (
uuid VARCHAR(20),
name VARCHAR(10),
age INT,
ts TIMESTAMP(3),
`partition` VARCHAR(20),
`region` VARCHAR(20),
PRIMARY KEY (uuid) NOT ENFORCED
) PARTITIONED BY (`partition`, `region`) WITH (
'connector' = 'hudi',
'path' = '/tmp/hudi-bucket-pruning-prefixed-key',
'index.type' = 'BUCKET'
);
INSERT INTO t1 VALUES
('id1', 'Danny', 23, TIMESTAMP '1970-01-01 00:00:01', 'par1', 'r1'),
('id2', 'Stephen', 33, TIMESTAMP '1970-01-01 00:00:02', 'par1', 'r2'),
('id3', 'Julian', 53, TIMESTAMP '1970-01-01 00:00:03', 'par2', 'r1'),
('id4', 'Fabian', 31, TIMESTAMP '1970-01-01 00:00:04', 'par2', 'r2'),
('id5', 'Sophia', 18, TIMESTAMP '1970-01-01 00:00:05', 'par3', 'r1'),
('id6', 'Emma', 20, TIMESTAMP '1970-01-01 00:00:06', 'par3', 'r2'),
('id7', 'Bob', 44, TIMESTAMP '1970-01-01 00:00:07', 'par4', 'r1'),
('id8', 'Han', 56, TIMESTAMP '1970-01-01 00:00:08', 'par4', 'r2');
SELECT * FROM t1; -- returns all 8 rows
SELECT * FROM t1 WHERE uuid = 'id1'; -- returns no rows
```
Reproducing test, added to `ITTestHoodieDataSource`:
```java
@ParameterizedTest
@MethodSource("tableTypeAndComplexKeyGenEncodingParams")
void
testBucketPruningWithSingleRecordKeyAndMultiplePartitionFields(HoodieTableType
tableType, ComplexKeyGenEncoding encoding) {
TableEnvironment tableEnv = batchTableEnv;
// a single record key with two partition fields uses the complex key
generator, which by default stores
// the record key as `uuid:<value>` (FIELD_PREFIXED); legacy version 8
tables may store the bare value (VALUE_ONLY)
String valueOnlyOptions = ",\n"
+ " '" + HoodieTableConfig.COMPLEX_KEYGEN_ENCODING.key() + "' = '"
+ ComplexKeyGenEncoding.VALUE_ONLY + "',\n"
+ " '" + FlinkOptions.WRITE_TABLE_VERSION.key() + "' = '" +
HoodieTableVersion.EIGHT.versionCode() + "'";
String hoodieTableDDL = "create table t1(\n"
+ " uuid varchar(20),\n"
+ " name varchar(10),\n"
+ " age int,\n"
+ " ts timestamp(3),\n"
+ " `partition` varchar(20),\n"
+ " `region` varchar(20),\n"
+ " PRIMARY KEY(uuid) NOT ENFORCED\n"
+ ")\n"
+ "PARTITIONED BY (`partition`, `region`)\n"
+ "with (\n"
+ " 'connector' = 'hudi',\n"
+ " 'path' = '" + tempFile.getAbsolutePath() + "',\n"
+ " '" + FlinkOptions.TABLE_TYPE.key() + "' = '" + tableType +
"',\n"
+ " '" + FlinkOptions.INDEX_TYPE.key() + "' = 'BUCKET'"
+ (encoding == ComplexKeyGenEncoding.VALUE_ONLY ? valueOnlyOptions :
"")
+ "\n)";
tableEnv.executeSql(hoodieTableDDL);
execInsertSql(tableEnv, "insert into t1 values\n"
+ "('id1','Danny',23,TIMESTAMP '1970-01-01 00:00:01','par1','r1'),\n"
+ "('id2','Stephen',33,TIMESTAMP '1970-01-01
00:00:02','par1','r2'),\n"
+ "('id3','Julian',53,TIMESTAMP '1970-01-01
00:00:03','par2','r1'),\n"
+ "('id4','Fabian',31,TIMESTAMP '1970-01-01
00:00:04','par2','r2'),\n"
+ "('id5','Sophia',18,TIMESTAMP '1970-01-01
00:00:05','par3','r1'),\n"
+ "('id6','Emma',20,TIMESTAMP '1970-01-01 00:00:06','par3','r2'),\n"
+ "('id7','Bob',44,TIMESTAMP '1970-01-01 00:00:07','par4','r1'),\n"
+ "('id8','Han',56,TIMESTAMP '1970-01-01 00:00:08','par4','r2')");
assertEquals(encoding,
HoodieTestUtils.createMetaClient(tempFile.getAbsolutePath()).getTableConfig().getComplexKeyGenEncoding().get());
List<Row> allRows = CollectionUtil.iterableToList(
() -> tableEnv.sqlQuery("select uuid from t1").execute().collect());
assertEquals(8, allRows.size());
// an equality filter on the record key must find the rows the
unfiltered scan finds
for (String uuid : Arrays.asList("id1", "id8")) {
List<Row> result = CollectionUtil.iterableToList(
() -> tableEnv.sqlQuery("select uuid from t1 where uuid = '" +
uuid + "'").execute().collect());
assertRowsEquals(result, "[+I[" + uuid + "]]");
}
}
/**
* Return test params => (HoodieTableType, ComplexKeyGenEncoding).
*/
private static Stream<Arguments> tableTypeAndComplexKeyGenEncodingParams()
{
Object[][] data =
new Object[][] {
{HoodieTableType.COPY_ON_WRITE,
ComplexKeyGenEncoding.FIELD_PREFIXED},
{HoodieTableType.COPY_ON_WRITE,
ComplexKeyGenEncoding.VALUE_ONLY},
{HoodieTableType.MERGE_ON_READ,
ComplexKeyGenEncoding.FIELD_PREFIXED},
{HoodieTableType.MERGE_ON_READ,
ComplexKeyGenEncoding.VALUE_ONLY}};
return Stream.of(data).map(Arguments::of);
}
```
On master, both `FIELD_PREFIXED` cases fail on the first key, and both
`VALUE_ONLY` cases pass:
```
[ERROR] Tests run: 4, Failures: 2, Errors: 0, Skipped: 0, Time elapsed:
35.60 s <<< FAILURE! -- in org.apache.hudi.table.ITTestHoodieDataSource
[ERROR]
org.apache.hudi.table.ITTestHoodieDataSource.testBucketPruningWithSingleRecordKeyAndMultiplePartitionFields(HoodieTableType,
ComplexKeyGenEncoding)[1] -- Time elapsed: 19.46 s <<< FAILURE!
java.lang.AssertionError:
Expected: is "[+I[id1]]"
but: was "[]"
[ERROR]
org.apache.hudi.table.ITTestHoodieDataSource.testBucketPruningWithSingleRecordKeyAndMultiplePartitionFields(HoodieTableType,
ComplexKeyGenEncoding)[3] -- Time elapsed: 3.481 s <<< FAILURE!
java.lang.AssertionError:
Expected: is "[+I[id1]]"
but: was "[]"
```
Cases `[1]` and `[3]` are COPY_ON_WRITE and MERGE_ON_READ with the default
`FIELD_PREFIXED` encoding. #20291 reports a different mismatch in the same
pruner, for `__null__` / `__empty__` values of compound keys.
### Environment
**Hudi version:** master (1.3.0-SNAPSHOT), `e9da226db3a9`
**Query engine:** Flink 1.20 (batch SQL)
**Relevant configs:** `index.type=BUCKET` with the default
`hoodie.bucket.index.num.buckets=4`, one `PRIMARY KEY` field, two `PARTITIONED
BY` fields, and the default
`hoodie.table.complex.keygenerator.encoding=FIELD_PREFIXED`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
