Skip to content

Full type test for 1127 #2

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
naivedogger wants to merge 5 commits into
base: main
Choose a base branch
from
Open

Full type test for 1127 #2

Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
f2ee55f
full-type test for UnionRead PrimaryKeyTable
Jun 18, 2025
ae61fa9
fix format problems&support partitioned tables
Jun 18, 2025
3f3d975
minor changes
Jun 18, 2025
e596183
Merge remote-tracking branch 'upstream/main' into full-type-test-for-…
Jun 18, 2025
2a6de99
fix
naivedogger Jun 19, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
269 changes: 267 additions & 2 deletions ...rc/test/java/com/alibaba/fluss/lake/paimon/flink/FlinkUnionReadPrimaryKeyTableITCase.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,7 @@
import com.alibaba.fluss.metadata.TableBucket;
import com.alibaba.fluss.metadata.TableDescriptor;
import com.alibaba.fluss.metadata.TablePath;
import com.alibaba.fluss.row.Decimal;
import com.alibaba.fluss.row.InternalRow;
import com.alibaba.fluss.row.TimestampLtz;
import com.alibaba.fluss.row.TimestampNtz;
Expand Down Expand Up @@ -73,7 +74,7 @@ void testPrimaryKeyTable(boolean isPartitioned) throws Exception {
Map<TableBucket, Long> bucketLogEndOffset = new HashMap<>();
long tableId = preparePkTable(t1, DEFAULT_BUCKET_NUM, isPartitioned, bucketLogEndOffset);

// wait unit records has has been synced
// wait unit records has been synced
waitUtilBucketSynced(t1, tableId, DEFAULT_BUCKET_NUM, isPartitioned);

// write records again
Expand Down Expand Up @@ -236,7 +237,7 @@ void testUnionReadWithTimeStamp() throws Exception {

writeRows(t1, rows, false);

// wait unit records has has been synced
// wait unit records has been synced
waitUtilBucketSynced(t1, tableId, 1, false);

// stop lake tiering service
Expand All @@ -259,6 +260,168 @@ void testUnionReadWithTimeStamp() throws Exception {
"[+I[1, 2023-10-25T12:01:13.182005Z, 2023-10-25T12:01:13.183006], +I[2, 2023-10-25T12:01:13.400007Z, 2023-10-25T12:01:13.501008]]");
}

@ParameterizedTest
@ValueSource(booleans = {false, true})
void testUnionReadFullType(Boolean isPartitioned) throws Exception {
// first of all, start tiering
JobClient jobClient = buildTieringJob(execEnv);

String tableName = "pk_table_full" + (isPartitioned ? "_partitioned" : "_non_partitioned");
TablePath t1 = TablePath.of(DEFAULT_DB, tableName);
Map<TableBucket, Long> bucketLogEndOffset = new HashMap<>();
long tableId = preparePKTableFullType(t1, 1, isPartitioned, bucketLogEndOffset);

// wait unit records has been synced
waitUtilBucketSynced(t1, tableId, 1, isPartitioned);

// stop lake tiering service
jobClient.cancel().get();

// write a row again
List<InternalRow> rows = new ArrayList<>();
if (isPartitioned) {
Map<Long, String> partitionNameById = waitUntilPartitions(t1);
for (String partition : partitionNameById.values()) {
rows =
Collections.singletonList(
row(
true,
(byte) 100,
(short) 200,
30,
400L,
500.1f,
600.0d,
"another_string_2",
Decimal.fromUnscaledLong(900, 5, 2),
Decimal.fromBigDecimal(
new java.math.BigDecimal(1000), 20, 0),
TimestampLtz.fromEpochMillis(1698235273400L, 7000),
TimestampNtz.fromMillis(1698235273501L, 8000),
new byte[] {5, 6, 7, 8},
partition));
writeRows(t1, rows, false);
}
} else {
rows =
Collections.singletonList(
row(
true,
(byte) 100,
(short) 200,
30,
400L,
500.1f,
600.0d,
"another_string_2",
Decimal.fromUnscaledLong(900, 5, 2),
Decimal.fromBigDecimal(new java.math.BigDecimal(1000), 20, 0),
TimestampLtz.fromEpochMillis(1698235273400L, 7000),
TimestampNtz.fromMillis(1698235273501L, 8000),
new byte[] {5, 6, 7, 8},
null));
writeRows(t1, rows, false);
}

// read paimon directly using $lake
TableResult tableResult =
batchTEnv.executeSql(String.format("select * from %s$lake", tableName));
List<String> paimonSnapshotRows =
CollectionUtil.iteratorToList(tableResult.collect()).stream()
.map(
row -> {
int userColumnCount = row.getArity() - 3;
Object[] fields = new Object[userColumnCount];
for (int i = 0; i < userColumnCount; i++) {
fields[i] = row.getField(i);
}
return Row.of(fields);
})
.map(Row::toString)
.sorted()
.collect(Collectors.toList());
if (isPartitioned) {
assertThat(paimonSnapshotRows.toString())
.isEqualTo(
"[+I[false, 1, 2, 3, 4, 5.1, 6.0, string, 0.09, 10, 2023-10-25T12:01:13.182005Z, 2023-10-25T12:01:13.183006, [1, 2, 3, 4], 2025], +I[false, 1, 2, 3, 4, 5.1, 6.0, string, 0.09, 10, 2023-10-25T12:01:13.182005Z, 2023-10-25T12:01:13.183006, [1, 2, 3, 4], 2026], +I[true, 10, 20, 30, 40, 50.1, 60.0, another_string, 0.90, 100, 2023-10-25T12:01:13.200005Z, 2023-10-25T12:01:13.201006, [1, 2, 3, 4], 2025], +I[true, 10, 20, 30, 40, 50.1, 60.0, another_string, 0.90, 100, 2023-10-25T12:01:13.200005Z, 2023-10-25T12:01:13.201006, [1, 2, 3, 4], 2026]]");
} else {
assertThat(paimonSnapshotRows.toString().replace("+U", "+I"))
.isEqualTo(
"[+I[false, 1, 2, 3, 4, 5.1, 6.0, string, 0.09, 10, 2023-10-25T12:01:13.182005Z, 2023-10-25T12:01:13.183006, [1, 2, 3, 4], null], +I[true, 10, 20, 30, 40, 50.1, 60.0, another_string, 0.90, 100, 2023-10-25T12:01:13.200005Z, 2023-10-25T12:01:13.201006, [1, 2, 3, 4], null]]");
}
// test point query with fluss
String queryFilterStr = "c4 = 30";
String partitionName =
isPartitioned ? waitUntilPartitions(t1).values().iterator().next() : null;
if (partitionName != null) {
queryFilterStr = queryFilterStr + " and c14= '" + partitionName + "'";
}

List<Row> expectedRows =
Arrays.asList(
Row.of(
false,
(byte) 1,
(short) 2,
3,
4L,
5.1f,
6.0d,
"string",
Decimal.fromUnscaledLong(9, 5, 2),
Decimal.fromBigDecimal(new java.math.BigDecimal(10), 20, 0),
TimestampLtz.fromEpochMillis(1698235273182L, 5000),
TimestampNtz.fromMillis(1698235273183L, 6000),
new byte[] {1, 2, 3, 4},
partitionName),
Row.of(
true,
(byte) 100,
(short) 200,
30,
400L,
500.1f,
600.0d,
"another_string_2",
Decimal.fromUnscaledLong(900, 5, 2),
Decimal.fromBigDecimal(new java.math.BigDecimal(1000), 20, 0),
TimestampLtz.fromEpochMillis(1698235273400L, 7000),
TimestampNtz.fromMillis(1698235273501L, 8000),
new byte[] {5, 6, 7, 8},
partitionName));
tableResult =
batchTEnv.executeSql(
String.format("select * from %s where %s", tableName, queryFilterStr));

List<String> paimonPointQueryRows = toSortedRows(tableResult);
List<String> expectedPointQueryRows =
expectedRows.stream()
.filter(
row -> {
boolean isMatch = row.getField(3).equals(30);
if (partitionName != null) {
isMatch = isMatch && row.getField(13).equals(partitionName);
}
return isMatch;
})
.map(Row::toString)
.sorted()
.collect(Collectors.toList());

assertThat(paimonPointQueryRows).isEqualTo(expectedPointQueryRows);
// now, query the result, it must union lake snapshot and log
List<String> result = toSortedRows(batchTEnv.executeSql("select * from " + tableName));
if (isPartitioned) {
assertThat(result.toString())
.isEqualTo(
"[+I[false, 1, 2, 3, 4, 5.1, 6.0, string, 0.09, 10, 2023-10-25T12:01:13.182005Z, 2023-10-25T12:01:13.183006, [1, 2, 3, 4], 2025], +I[false, 1, 2, 3, 4, 5.1, 6.0, string, 0.09, 10, 2023-10-25T12:01:13.182005Z, 2023-10-25T12:01:13.183006, [1, 2, 3, 4], 2026], +I[true, 100, 200, 30, 400, 500.1, 600.0, another_string_2, 9.00, 1000, 2023-10-25T12:01:13.400007Z, 2023-10-25T12:01:13.501008, [5, 6, 7, 8], 2025], +I[true, 100, 200, 30, 400, 500.1, 600.0, another_string_2, 9.00, 1000, 2023-10-25T12:01:13.400007Z, 2023-10-25T12:01:13.501008, [5, 6, 7, 8], 2026]]");
} else {
assertThat(result.toString().replace("+U", "+I"))
.isEqualTo(
"[+I[false, 1, 2, 3, 4, 5.1, 6.0, string, 0.09, 10, 2023-10-25T12:01:13.182005Z, 2023-10-25T12:01:13.183006, [1, 2, 3, 4], null], +I[true, 100, 200, 30, 400, 500.1, 600.0, another_string_2, 9.00, 1000, 2023-10-25T12:01:13.400007Z, 2023-10-25T12:01:13.501008, [5, 6, 7, 8], null]]");
}
}

private List<Row> paddingPartition(TablePath tablePath, List<Row> rows) {
List<Row> paddingPartitionRows = new ArrayList<>();
for (String partition : waitUntilPartitions(tablePath).values()) {
Expand Down Expand Up @@ -309,6 +472,36 @@ private long preparePkTable(
return tableId;
}

private long preparePKTableFullType(
TablePath tablePath,
int bucketNum,
boolean isPartitioned,
Map<TableBucket, Long> bucketLogEndOffset)
throws Exception {
long tableId = createPkTableFullType(tablePath, bucketNum, isPartitioned);
if (isPartitioned) {
Map<Long, String> partitionNameById = waitUntilPartitions(tablePath);
for (String partition : partitionNameById.values()) {
for (int i = 0; i < 2; i++) {
List<InternalRow> rows = generateKvRowsFullType(partition);
// write records
writeRows(tablePath, rows, false);
}
}
for (Long partitionId : partitionNameById.keySet()) {
bucketLogEndOffset.putAll(getBucketLogEndOffset(tableId, bucketNum, partitionId));
}
} else {
for (int i = 0; i < 2; i++) {
List<InternalRow> rows = generateKvRowsFullType(null);
// write records
writeRows(tablePath, rows, false);
}
bucketLogEndOffset.putAll(getBucketLogEndOffset(tableId, bucketNum, null));
}
return tableId;
}

private void writeRowsToPkTable(
TablePath tablePath,
long tableId,
Expand Down Expand Up @@ -397,6 +590,44 @@ protected long createPkTable(TablePath tablePath, int bucketNum, boolean isParti
return createTable(tablePath, tableBuilder.build());
}

protected long createPkTableFullType(TablePath tablePath, int bucketNum, boolean isPartitioned)
throws Exception {
Schema.Builder schemaBuilder =
Schema.newBuilder()
.column("c1", DataTypes.BOOLEAN())
.column("c2", DataTypes.TINYINT())
.column("c3", DataTypes.SMALLINT())
.column("c4", DataTypes.INT())
.column("c5", DataTypes.BIGINT())
.column("c6", DataTypes.FLOAT())
.column("c7", DataTypes.DOUBLE())
.column("c8", DataTypes.STRING())
.column("c9", DataTypes.DECIMAL(5, 2))
.column("c10", DataTypes.DECIMAL(20, 0))
.column("c11", DataTypes.TIMESTAMP_LTZ(6))
.column("c12", DataTypes.TIMESTAMP(6))
.column("c13", DataTypes.BINARY(4))
.column("c14", DataTypes.STRING());

TableDescriptor.Builder tableBuilder =
TableDescriptor.builder()
.distributedBy(bucketNum)
.property(ConfigOptions.TABLE_DATALAKE_ENABLED.key(), "true")
.property(ConfigOptions.TABLE_DATALAKE_FRESHNESS, Duration.ofMillis(500));

if (isPartitioned) {
tableBuilder.property(ConfigOptions.TABLE_AUTO_PARTITION_ENABLED, true);
tableBuilder.partitionedBy("c14");
schemaBuilder.primaryKey("c4", "c14");
tableBuilder.property(
ConfigOptions.TABLE_AUTO_PARTITION_TIME_UNIT, AutoPartitionTimeUnit.YEAR);
} else {
schemaBuilder.primaryKey("c4");
}
tableBuilder.schema(schemaBuilder.build());
return createTable(tablePath, tableBuilder.build());
}

private List<InternalRow> generateKvRows(@Nullable String partition) {
if (partition == null) {
return Arrays.asList(
Expand All @@ -414,4 +645,38 @@ private List<InternalRow> generateKvRows(@Nullable String partition) {
row("f2222", 2222, "v2222", partition));
}
}

private List<InternalRow> generateKvRowsFullType(@Nullable String partition) {
return Arrays.asList(
row(
false,
(byte) 1,
(short) 2,
3,
4L,
5.1f,
6.0d,
"string",
Decimal.fromUnscaledLong(9, 5, 2),
Decimal.fromBigDecimal(new java.math.BigDecimal(10), 20, 0),
TimestampLtz.fromEpochMillis(1698235273182L, 5000),
TimestampNtz.fromMillis(1698235273183L, 6000),
new byte[] {1, 2, 3, 4},
partition),
row(
true,
(byte) 10,
(short) 20,
30,
40L,
50.1f,
60.0d,
"another_string",
Decimal.fromUnscaledLong(90, 5, 2),
Decimal.fromBigDecimal(new java.math.BigDecimal(100), 20, 0),
TimestampLtz.fromEpochMillis(1698235273200L, 5000),
TimestampNtz.fromMillis(1698235273201L, 6000),
new byte[] {1, 2, 3, 4},
partition));
}
}
Loading